Merge pull request #31146 from NeQuissimus/kafka_updates

Kafka: Update + Tests

nixos/release.nix

···
   tests.plasma5 = callTest tests/plasma5.nix {};
   tests.keymap = callSubTests tests/keymap.nix {};
   tests.initrdNetwork = callTest tests/initrd-network.nix {};
+  tests.kafka_0_9 = callTest tests/kafka_0_9.nix {};
+  tests.kafka_0_10 = callTest tests/kafka_0_10.nix {};
+  tests.kafka_0_11 = callTest tests/kafka_0_11.nix {};
+  tests.kafka_1_0 = callTest tests/kafka_1_0.nix {};
   tests.kernel-copperhead = callTest tests/kernel-copperhead.nix {};
   tests.kernel-latest = callTest tests/kernel-latest.nix {};
   tests.kernel-lts = callTest tests/kernel-lts.nix {};
···
   tests.wordpress = callTest tests/wordpress.nix {};
   tests.xfce = callTest tests/xfce.nix {};
   tests.xmonad = callTest tests/xmonad.nix {};
-
+  tests.zookeeper = callTest tests/zookeeper.nix {};
 
   /* Build a bunch of typical closures so that Hydra can keep track of
      the evolution of closure sizes. */

nixos/tests/kafka_0_10.nix

···
+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'");
+  '';
+})

nixos/tests/kafka_0_11.nix

···
+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'");
+  '';
+})

nixos/tests/kafka_0_9.nix

···
+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'");
+  '';
+})

nixos/tests/kafka_1_0.nix

···
+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'");
+  '';
+})

nixos/tests/zookeeper.nix

···
+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");
+  '';
+})

pkgs/servers/apache-kafka/default.nix

···
 { stdenv, fetchurl, jre, makeWrapper, bash,
-  majorVersion ? "0.9" }:
+  majorVersion ? "1.0" }:
 
 let
   versionMap = {
-    "0.8" = { kafkaVersion = "0.8.2.2";
-              scalaVersion = "2.10";
-              sha256 = "1azccf1k0nr8y1sfpjgqf9swyp87ypvgva68ci4kczwcx1z9d89v";
-            };
-    "0.9" = { kafkaVersion = "0.9.0.1";
-              scalaVersion = "2.11";
-              sha256 = "0ykcjv5dz9i5bws9my2d60pww1g9v2p2nqr67h0i2xrjm7az8a6v";
-            };
-    "0.10" = { kafkaVersion = "0.10.2.0";
-               scalaVersion = "2.12";
-               sha256 = "0py43s6zv8z7wr2lk8403k07xxckl11gla3vs4gr99lixc4whis1";
-             };
+    "0.9" = {
+      kafkaVersion = "0.9.0.1";
+      scalaVersion = "2.11";
+      sha256 = "0ykcjv5dz9i5bws9my2d60pww1g9v2p2nqr67h0i2xrjm7az8a6v";
+    };
+    "0.10" = {
+      kafkaVersion = "0.10.2.1";
+      scalaVersion = "2.12";
+      sha256 = "0iszr6r0n9yjgq7kcp1hf00fg754m86gs4jzqc18542an94b88z5";
+    };
+    "0.11" = {
+      kafkaVersion = "0.11.0.1";
+      scalaVersion = "2.12";
+      sha256 = "1wj639h95aq5n132fq1rbyzqh5rsa4mlhbg3c5mszqglnzdz4xn7";
+    };
+    "1.0" = {
+      kafkaVersion = "1.0.0";
+      scalaVersion = "2.12";
+      sha256 = "1cs4nmp39m99gqjpy5klsffqksc0h9pz514jkq99qb95a83x1cfm";
+    };
   };
 in
 

pkgs/top-level/all-packages.nix

···
 
   apacheAnt = callPackage ../development/tools/build-managers/apache-ant { };
 
-  apacheKafka = apacheKafka_0_10;
-  apacheKafka_0_8 = callPackage ../servers/apache-kafka { majorVersion = "0.8"; };
+  apacheKafka = apacheKafka_1_0;
   apacheKafka_0_9 = callPackage ../servers/apache-kafka { majorVersion = "0.9"; };
   apacheKafka_0_10 = callPackage ../servers/apache-kafka { majorVersion = "0.10"; };
+  apacheKafka_0_11 = callPackage ../servers/apache-kafka { majorVersion = "0.11"; };
+  apacheKafka_1_0 = callPackage ../servers/apache-kafka { majorVersion = "1.0"; };
 
   kt = callPackage ../tools/misc/kt {};