Set HighAvailabilityTaskAssignor as default in streams_upgrade_test.py
Generalize the verification in the upgrade test so that it
does not rely on the task assignor's behavior.
Reviewers: A. Sophie Blee-Goldman John Roesler
@@ -303,13 +303,9 @@ class StreamsUpgradeTest(Test):
self.driver = StreamsSmokeTestDriverService(self.test_context, self.kafka)
self.driver.disable_auto_terminate()
- # TODO KIP-441: consider rewriting the test for HighAvailabilityTaskAssignor
self.processor1 = StreamsUpgradeTestJobRunnerService(self.test_context, self.kafka)
- self.processor1.set_config("internal.task.assignor.class", "org.apache.kafka.streams.processor.internals.assignment.StickyTaskAssignor")
self.processor2 = StreamsUpgradeTestJobRunnerService(self.test_context, self.kafka)
- self.processor2.set_config("internal.task.assignor.class", "org.apache.kafka.streams.processor.internals.assignment.StickyTaskAssignor")
self.processor3 = StreamsUpgradeTestJobRunnerService(self.test_context, self.kafka)
- self.processor3.set_config("internal.task.assignor.class", "org.apache.kafka.streams.processor.internals.assignment.StickyTaskAssignor")
self.driver.start()
self.start_all_nodes_with("") # run with TRUNK
@@ -532,18 +528,20 @@ class StreamsUpgradeTest(Test):
log_monitor.wait_until("Sent a version 8 subscription and got version 7 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 7 and trigger new rebalance.",
timeout_sec=60,
err_msg="Could not detect 'successful version probing' at upgrading node " + str(node.account))
- else:
- log_monitor.wait_until("Sent a version 8 subscription and got version 7 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 8 and trigger new rebalance.",
+ log_monitor.wait_until("Detected that the assignor requested a rebalance. Rejoining the consumer group to trigger a new rebalance.",
timeout_sec=60,
- err_msg="Could not detect 'successful version probing with upgraded leader' at upgrading node " + str(node.account))
+ err_msg="Could not detect 'Triggering new rebalance' at upgrading node " + str(node.account))
+ else:
first_other_monitor.wait_until("Sent a version 7 subscription and group.s latest commonly supported version is 8 (successful version probing and end of rolling upgrade). Upgrading subscription metadata version to 8 for next rebalance.",
timeout_sec=60,
err_msg="Never saw output 'Upgrade metadata to version 8' on" + str(first_other_node.account))
+ first_other_monitor.wait_until("Detected that the assignor requested a rebalance. Rejoining the consumer group to trigger a new rebalance.",
+ timeout_sec=60,
+ err_msg="Could not detect 'Triggering new rebalance' at upgrading node " + str(node.account))
second_other_monitor.wait_until("Sent a version 7 subscription and group.s latest commonly supported version is 8 (successful version probing and end of rolling upgrade). Upgrading subscription metadata version to 8 for next rebalance.",
timeout_sec=60,
err_msg="Never saw output 'Upgrade metadata to version 8' on" + str(second_other_node.account))
-
- log_monitor.wait_until("Detected that the assignor requested a rebalance. Rejoining the consumer group to trigger a new rebalance.",
+ second_other_monitor.wait_until("Detected that the assignor requested a rebalance. Rejoining the consumer group to trigger a new rebalance.",
timeout_sec=60,
err_msg="Could not detect 'Triggering new rebalance' at upgrading node " + str(node.account))
