From 0080b24694f56500138c3511791631fc32dafd6e Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Sun, 19 May 2019 19:59:51 +0200 Subject: [PATCH] MINOR: fix Streams version-probing system test --- .../streams/tests/StreamsUpgradeTest.java | 20 ++++++++----------- .../tests/streams/streams_upgrade_test.py | 7 +++---- 2 files changed, 11 insertions(+), 16 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java b/streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java index 33e9b9771c245..27bee810d85fd 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java @@ -47,7 +47,6 @@ import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -90,16 +89,13 @@ public static void main(final String[] args) throws Exception { final KafkaStreams streams = new KafkaStreams(builder.build(), config, kafkaClientSupplier); streams.start(); - Runtime.getRuntime().addShutdownHook(new Thread() { - @Override - public void run() { - System.out.println("closing Kafka Streams instance"); - System.out.flush(); - streams.close(); - System.out.println("UPGRADE-TEST-CLIENT-CLOSED"); - System.out.flush(); - } - }); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("closing Kafka Streams instance"); + System.out.flush(); + streams.close(); + System.out.println("UPGRADE-TEST-CLIENT-CLOSED"); + System.out.flush(); + })); } private static class FutureKafkaClientSupplier extends DefaultKafkaClientSupplier { @@ -168,7 +164,7 @@ public void onAssignment(final PartitionAssignor.Assignment assignment) { assignment.userData().putInt(0, AssignmentInfo.LATEST_SUPPORTED_VERSION)); final List partitions = new ArrayList<>(assignment.partitions()); - Collections.sort(partitions, PARTITION_COMPARATOR); + partitions.sort(PARTITION_COMPARATOR); // version 1 field final Map> activeTasks = new HashMap<>(); diff --git a/tests/kafkatest/tests/streams/streams_upgrade_test.py b/tests/kafkatest/tests/streams/streams_upgrade_test.py index 8b629775237a9..4cb8bb63a632d 100644 --- a/tests/kafkatest/tests/streams/streams_upgrade_test.py +++ b/tests/kafkatest/tests/streams/streams_upgrade_test.py @@ -70,6 +70,7 @@ """ + class StreamsUpgradeTest(Test): """ Test upgrading Kafka Streams (all version combination) @@ -348,7 +349,7 @@ def update_leader(self): while retries > 0: for p in self.processors: found = list(p.node.account.ssh_capture("grep \"Finished assignment for group\" %s" % p.LOG_FILE, allow_fail=True)) - if len(found) == self.leader_counter[p] + 1: + if len(found) >= self.leader_counter[p] + 1: if self.leader is not None: raise Exception("Could not uniquely identify leader") self.leader = p @@ -403,8 +404,7 @@ def start_all_nodes_with(self, version): timeout_sec=60, err_msg="Never saw output '%s' on " % self.processed_msg + str(node2.account)) - - # start third with + # start third with self.prepare_for(self.processor3, version) node3 = self.processor3.node with node1.account.monitor_log(self.processor1.STDOUT_FILE) as first_monitor: @@ -425,7 +425,6 @@ def start_all_nodes_with(self, version): timeout_sec=60, err_msg="Never saw output '%s' on " % self.processed_msg + str(node3.account)) - @staticmethod def prepare_for(processor, version): processor.node.account.ssh("rm -rf " + processor.PERSISTENT_ROOT, allow_fail=False)