nixos/tests/kafka: fix and refactor tests

- refactor into single file for all versions
- improve timing, prevent non-deterministic failures
- fix tests for i686-linux

6 files changed, 69 insertions, 196 deletions

diff --git a/nixos/release.nix b/nixos/release.nix
index ae70b535a5e2..57fc245e57dd 100644
--- a/nixos/release.nix
+++ b/nixos/release.nix
@@ -313,10 +313,7 @@ in rec {
   tests.plotinus = callTest tests/plotinus.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.kafka = callSubTests tests/kafka.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 {};
diff --git a/nixos/tests/kafka.nix b/nixos/tests/kafka.nix
new file mode 100644
index 000000000000..6209b4b8e2b7
--- /dev/null
+++ b/nixos/tests/kafka.nix
@@ -0,0 +1,68 @@
+{ system ? builtins.currentSystem }:
+with import ../lib/testing.nix { inherit system; };
+with pkgs.lib;
+
+let
+  makeKafkaTest = name: kafkaPackage: (makeTest {
+    inherit name;
+    meta = with pkgs.stdenv.lib.maintainers; {
+      maintainers = [ nequissimus ];
+    };
+
+    nodes = {
+      zookeeper1 = { config, ... }: {
+        services.zookeeper = {
+          enable = true;
+        };
+
+        networking.firewall.allowedTCPPorts = [ 2181 ];
+        virtualisation.memorySize = 1024;
+      };
+      kafka = { config, ... }: {
+        services.apache-kafka = {
+          enable = true;
+          extraProperties = ''
+            offsets.topic.replication.factor = 1
+            zookeeper.session.timeout.ms = 600000
+          '';
+          package = kafkaPackage;
+          zookeeper = "zookeeper1:2181";
+          # These are the default options, but UseCompressedOops doesn't work with 32bit JVM
+          jvmOptions = [
+            "-server" "-Xmx1G" "-Xms1G" "-XX:+UseParNewGC" "-XX:+UseConcMarkSweepGC" "-XX:+CMSClassUnloadingEnabled"
+            "-XX:+CMSScavengeBeforeRemark" "-XX:+DisableExplicitGC" "-Djava.awt.headless=true" "-Djava.net.preferIPv4Stack=true"
+          ] ++ optionals (! pkgs.stdenv.isi686 ) [ "-XX:+UseCompressedOops" ];
+        };
+
+        networking.firewall.allowedTCPPorts = [ 9092 ];
+        # i686 tests: qemu-system-i386 can simulate max 2047MB RAM (not 2048)
+        virtualisation.memorySize = 2047; 
+      };
+    };
+
+    testScript = ''
+      startAll;
+
+      $zookeeper1->waitForUnit("default.target");
+      $zookeeper1->waitForUnit("zookeeper.service");
+      $zookeeper1->waitForOpenPort(2181);
+
+      $kafka->waitForUnit("default.target");
+      $kafka->waitForUnit("apache-kafka.service");
+      $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");
+    '' + (if name == "kafka_0_9" then ''
+      $kafka->mustSucceed("${kafkaPackage}/bin/kafka-console-consumer.sh --zookeeper zookeeper1:2181 --topic testtopic --from-beginning --max-messages 1 | grep 'test 1'");
+    '' else ''
+      $kafka->mustSucceed("${kafkaPackage}/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopic --from-beginning --max-messages 1 | grep 'test 1'");
+    '');
+  });
+
+in with pkgs; {
+  kafka_0_9  = makeKafkaTest "kafka_0_9"  apacheKafka_0_9;
+  kafka_0_10 = makeKafkaTest "kafka_0_10" apacheKafka_0_10;
+  kafka_0_11 = makeKafkaTest "kafka_0_11" apacheKafka_0_11;
+  kafka_1_0  = makeKafkaTest "kafka_1_0"  apacheKafka_1_0;
+}
diff --git a/nixos/tests/kafka_0_10.nix b/nixos/tests/kafka_0_10.nix
deleted file mode 100644
index 6e7820f64bc4..000000000000
--- a/nixos/tests/kafka_0_10.nix
+++ /dev/null
@@ -1,48 +0,0 @@
-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
deleted file mode 100644
index 39f9c36bb229..000000000000
--- a/nixos/tests/kafka_0_11.nix
+++ /dev/null
@@ -1,48 +0,0 @@
-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
deleted file mode 100644
index fee82aba2bda..000000000000
--- a/nixos/tests/kafka_0_9.nix
+++ /dev/null
@@ -1,48 +0,0 @@
-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
deleted file mode 100644
index 936840dbcfdc..000000000000
--- a/nixos/tests/kafka_1_0.nix
+++ /dev/null
@@ -1,48 +0,0 @@
-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'");
-  '';
-})