Merge pull request #31146 from NeQuissimus/kafka_updates

Kafka: Update + Tests

5 files changed, 220 insertions, 0 deletions

diff --git a/nixos/tests/kafka_0_10.nix b/nixos/tests/kafka_0_10.nix
new file mode 100644
index 00000000000..6e7820f64bc
--- /dev/null
+++ b/nixos/tests/kafka_0_10.nix
@@ -0,0 +1,48 @@
+import ./make-test.nix ({ pkgs, lib, ... } :
+let
+  kafkaPackage = pkgs.apacheKafka_0_10;
+in {
+  name = "kafka_0_10";
+  meta = with pkgs.stdenv.lib.maintainers; {
+    maintainers = [ nequissimus ];
+  };
+
+  nodes = {
+    zookeeper1 = { config, ... }: {
+      services.zookeeper = {
+        enable = true;
+      };
+
+      networking.firewall.allowedTCPPorts = [ 2181 ];
+    };
+    kafka = { config, ... }: {
+      services.apache-kafka = {
+        enable = true;
+        extraProperties = ''
+          offsets.topic.replication.factor = 1
+        '';
+        package = kafkaPackage;
+        zookeeper = "zookeeper1:2181";
+      };
+
+      networking.firewall.allowedTCPPorts = [ 9092 ];
+      virtualisation.memorySize = 2048;
+    };
+  };
+
+  testScript = ''
+    startAll;
+
+    $zookeeper1->waitForUnit("zookeeper");
+    $zookeeper1->waitForUnit("network.target");
+    $zookeeper1->waitForOpenPort(2181);
+
+    $kafka->waitForUnit("apache-kafka");
+    $kafka->waitForUnit("network.target");
+    $kafka->waitForOpenPort(9092);
+
+    $kafka->waitUntilSucceeds("${kafkaPackage}/bin/kafka-topics.sh --create --zookeeper zookeeper1:2181 --partitions 1 --replication-factor 1 --topic testtopic");
+    $kafka->mustSucceed("echo 'test 1' | ${kafkaPackage}/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopic");
+    $kafka->mustSucceed("${kafkaPackage}/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopic --from-beginning --max-messages 1 | grep 'test 1'");
+  '';
+})
diff --git a/nixos/tests/kafka_0_11.nix b/nixos/tests/kafka_0_11.nix
new file mode 100644
index 00000000000..39f9c36bb22
--- /dev/null
+++ b/nixos/tests/kafka_0_11.nix
@@ -0,0 +1,48 @@
+import ./make-test.nix ({ pkgs, lib, ... } :
+let
+  kafkaPackage = pkgs.apacheKafka_0_11;
+in {
+  name = "kafka_0_11";
+  meta = with pkgs.stdenv.lib.maintainers; {
+    maintainers = [ nequissimus ];
+  };
+
+  nodes = {
+    zookeeper1 = { config, ... }: {
+      services.zookeeper = {
+        enable = true;
+      };
+
+      networking.firewall.allowedTCPPorts = [ 2181 ];
+    };
+    kafka = { config, ... }: {
+      services.apache-kafka = {
+        enable = true;
+        extraProperties = ''
+          offsets.topic.replication.factor = 1
+        '';
+        package = kafkaPackage;
+        zookeeper = "zookeeper1:2181";
+      };
+
+      networking.firewall.allowedTCPPorts = [ 9092 ];
+      virtualisation.memorySize = 2048;
+    };
+  };
+
+  testScript = ''
+    startAll;
+
+    $zookeeper1->waitForUnit("zookeeper");
+    $zookeeper1->waitForUnit("network.target");
+    $zookeeper1->waitForOpenPort(2181);
+
+    $kafka->waitForUnit("apache-kafka");
+    $kafka->waitForUnit("network.target");
+    $kafka->waitForOpenPort(9092);
+
+    $kafka->waitUntilSucceeds("${kafkaPackage}/bin/kafka-topics.sh --create --zookeeper zookeeper1:2181 --partitions 1 --replication-factor 1 --topic testtopic");
+    $kafka->mustSucceed("echo 'test 1' | ${kafkaPackage}/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopic");
+    $kafka->mustSucceed("${kafkaPackage}/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopic --from-beginning --max-messages 1 | grep 'test 1'");
+  '';
+})
diff --git a/nixos/tests/kafka_0_9.nix b/nixos/tests/kafka_0_9.nix
new file mode 100644
index 00000000000..fee82aba2bd
--- /dev/null
+++ b/nixos/tests/kafka_0_9.nix
@@ -0,0 +1,48 @@
+import ./make-test.nix ({ pkgs, lib, ... } :
+let
+  kafkaPackage = pkgs.apacheKafka_0_9;
+in {
+  name = "kafka_0_9";
+  meta = with pkgs.stdenv.lib.maintainers; {
+    maintainers = [ nequissimus ];
+  };
+
+  nodes = {
+    zookeeper1 = { config, ... }: {
+      services.zookeeper = {
+        enable = true;
+      };
+
+      networking.firewall.allowedTCPPorts = [ 2181 ];
+    };
+    kafka = { config, ... }: {
+      services.apache-kafka = {
+        enable = true;
+        extraProperties = ''
+          offsets.topic.replication.factor = 1
+        '';
+        package = kafkaPackage;
+        zookeeper = "zookeeper1:2181";
+      };
+
+      networking.firewall.allowedTCPPorts = [ 9092 ];
+      virtualisation.memorySize = 2048;
+    };
+  };
+
+  testScript = ''
+    startAll;
+
+    $zookeeper1->waitForUnit("zookeeper");
+    $zookeeper1->waitForUnit("network.target");
+    $zookeeper1->waitForOpenPort(2181);
+
+    $kafka->waitForUnit("apache-kafka");
+    $kafka->waitForUnit("network.target");
+    $kafka->waitForOpenPort(9092);
+
+    $kafka->waitUntilSucceeds("${kafkaPackage}/bin/kafka-topics.sh --create --zookeeper zookeeper1:2181 --partitions 1 --replication-factor 1 --topic testtopic");
+    $kafka->mustSucceed("echo 'test 1' | ${kafkaPackage}/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopic");
+    $kafka->mustSucceed("${kafkaPackage}/bin/kafka-console-consumer.sh --zookeeper zookeeper1:2181 --topic testtopic --from-beginning --max-messages 1 | grep 'test 1'");
+  '';
+})
diff --git a/nixos/tests/kafka_1_0.nix b/nixos/tests/kafka_1_0.nix
new file mode 100644
index 00000000000..936840dbcfd
--- /dev/null
+++ b/nixos/tests/kafka_1_0.nix
@@ -0,0 +1,48 @@
+import ./make-test.nix ({ pkgs, lib, ... } :
+let
+  kafkaPackage = pkgs.apacheKafka_1_0;
+in {
+  name = "kafka_1_0";
+  meta = with pkgs.stdenv.lib.maintainers; {
+    maintainers = [ nequissimus ];
+  };
+
+  nodes = {
+    zookeeper1 = { config, ... }: {
+      services.zookeeper = {
+        enable = true;
+      };
+
+      networking.firewall.allowedTCPPorts = [ 2181 ];
+    };
+    kafka = { config, ... }: {
+      services.apache-kafka = {
+        enable = true;
+        extraProperties = ''
+          offsets.topic.replication.factor = 1
+        '';
+        package = kafkaPackage;
+        zookeeper = "zookeeper1:2181";
+      };
+
+      networking.firewall.allowedTCPPorts = [ 9092 ];
+      virtualisation.memorySize = 2048;
+    };
+  };
+
+  testScript = ''
+    startAll;
+
+    $zookeeper1->waitForUnit("zookeeper");
+    $zookeeper1->waitForUnit("network.target");
+    $zookeeper1->waitForOpenPort(2181);
+
+    $kafka->waitForUnit("apache-kafka");
+    $kafka->waitForUnit("network.target");
+    $kafka->waitForOpenPort(9092);
+
+    $kafka->waitUntilSucceeds("${kafkaPackage}/bin/kafka-topics.sh --create --zookeeper zookeeper1:2181 --partitions 1 --replication-factor 1 --topic testtopic");
+    $kafka->mustSucceed("echo 'test 1' | ${kafkaPackage}/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopic");
+    $kafka->mustSucceed("${kafkaPackage}/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopic --from-beginning --max-messages 1 | grep 'test 1'");
+  '';
+})
diff --git a/nixos/tests/zookeeper.nix b/nixos/tests/zookeeper.nix
new file mode 100644
index 00000000000..d247654adad
--- /dev/null
+++ b/nixos/tests/zookeeper.nix
@@ -0,0 +1,28 @@
+import ./make-test.nix ({ pkgs, ...} : {
+  name = "zookeeper";
+  meta = with pkgs.stdenv.lib.maintainers; {
+    maintainers = [ nequissimus ];
+  };
+
+  nodes = {
+    server = { pkgs, config, ... }: {
+      services.zookeeper = {
+        enable = true;
+      };
+
+      networking.firewall.allowedTCPPorts = [ 2181 ];
+    };
+  };
+
+  testScript = ''
+    startAll;
+
+    $server->waitForUnit("zookeeper");
+    $server->waitForUnit("network.target");
+    $server->waitForOpenPort(2181);
+
+    $server->waitUntilSucceeds("${pkgs.zookeeper}/bin/zkCli.sh -server localhost:2181 create /foo bar");
+    $server->waitUntilSucceeds("${pkgs.zookeeper}/bin/zkCli.sh -server localhost:2181 set /foo hello");
+    $server->waitUntilSucceeds("${pkgs.zookeeper}/bin/zkCli.sh -server localhost:2181 get /foo | grep hello");
+  '';
+})