fqaiser94 commented on code in PR #13656: URL: https://github.com/apache/kafka/pull/13656#discussion_r1190418116
########## tests/kafkatest/tests/streams/streams_upgrade_test.py: ########## @@ -236,6 +237,96 @@ def test_rolling_upgrade_with_2_bounces(self, from_version, to_version): self.stop_and_await() + @cluster(num_nodes=6) + @matrix(from_version=table_agg_versions, to_version=[str(DEV_VERSION)]) + def test_rolling_upgrade_for_table_agg(self, from_version, to_version): + """ + This test verifies that the cluster successfully upgrades despite changes in the table + repartition topic format. + + Starts 3 KafkaStreams instances with version <from_version> and upgrades one-by-one to <to_version> + """ + + extra_properties = {'test.run_table_agg': 'true'} + + self.set_up_services() + + self.driver.start() + + # encoding different target values for different versions + # - old version: value=A + # - new version with `upgrade_from` flag set: value=B + # - new version w/o `upgrade_from` set set: value=C + + extra_properties = extra_properties.copy() + extra_properties['test.agg_produce_value'] = 'A' + extra_properties['test.expected_agg_values'] = 'A' + self.start_all_nodes_with(from_version, extra_properties) + + counter = 1 + random.seed() + + # rolling bounce + random.shuffle(self.processors) + p3 = self.processors[-1] + for p in self.processors: + p.CLEAN_NODE_ENABLED = False + + # bounce two instances to new version (verifies that new version can process records + # written by old version) + extra_properties = extra_properties.copy() + extra_properties['test.agg_produce_value'] = 'B' + extra_properties['test.expected_agg_values'] = 'A,B' + for p in self.processors[:-1]: + self.do_stop_start_bounce(p, from_version[:-2], to_version, counter, extra_properties) + counter = counter + 1 + + # bounce remaining instance on old version (just for verification purposes, to verify that + # instance on old version can process records written by new version) + extra_properties = extra_properties.copy() + extra_properties['test.agg_produce_value'] = 'A' + extra_properties['test.expected_agg_values'] = 'A,B' + self.do_stop_start_bounce(p3, None, from_version, counter, extra_properties) + counter = counter + 1 + + self.wait_for_table_agg_success('A,B') Review Comment: ```suggestion # bounce remaining instance on old version to produce a new unique value extra_properties = extra_properties.copy() extra_properties['test.agg_produce_value'] = 'C' extra_properties['test.expected_agg_values'] = 'A,B,C' self.do_stop_start_bounce(p3, None, from_version, counter, extra_properties) counter = counter + 1 # verify that new version can process records from old version self.wait_for_table_agg_success('A,B,C') ``` I think it might be better to have this old (not-upgraded) instance start producing a new value `'C'` when we bounce it here. That way, we can assert using `self.wait_for_table_agg_success('A,B,C')` and be sure that the two upgraded instances have successfully processed messages from the old (not-upgraded) instance as well. ![image](https://github.com/apache/kafka/assets/20507243/0f703375-d2b9-44e8-85f1-94a7dc2a6398) (Note, if you accept this change, you will need to make changes below here to produce new unique values like D,E,F, etc.) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org