chia7712 commented on code in PR #23075:
URL: https://github.com/apache/kafka/pull/23075#discussion_r3842694815
##########
tests/kafkatest/tests/streams/streams_application_upgrade_test.py:
##########
@@ -33,6 +33,16 @@
str(LATEST_4_0), str(LATEST_4_1), str(LATEST_4_2),
str(LATEST_4_3)]
+# The headers-aware suppress buffer (KAFKA-20413) is only on trunk, so 4.3 is
the newest
+# release that still writes plain V3 suppress-changelog records even with this
config set.
+DSL_STORE_FORMAT_CONFIG = "dsl.store.format"
+DSL_STORE_FORMAT_HEADERS = "HEADERS"
+SUPPRESS_HEADERS_OLD_VERSION = str(LATEST_4_3)
+
+# InMemoryTimeOrderedKeyValueChangeBuffer throws this when it cannot make
sense of a
+# suppress-changelog record while restoring.
+INVALID_CHANGELOG_RECORD_MSG = "Restoring apparently invalid changelog record"
Review Comment:
Should we add a comment to InMemoryTimeOrderedKeyValueChangeBuffer​
reminding developers not to change this message?
```java
} else {
throw new IllegalArgumentException("Restoring apparently
invalid changelog record: " + record);
}
```
##########
tests/kafkatest/tests/streams/streams_application_upgrade_test.py:
##########
@@ -63,8 +73,48 @@ def test_app_upgrade(self, from_version, bounce_type,
metadata_quorum):
"""
Starts 3 KafkaStreams instances with <old_version>, and upgrades
one-by-one to <new_version>
"""
+ self._run_app_transition(from_version, str(DEV_VERSION), bounce_type)
+
+ @cluster(num_nodes=9)
+ @matrix(direction=["upgrade", "downgrade"],
metadata_quorum=[quorum.combined_kraft])
+ def test_suppress_headers_app_transition(self, direction, metadata_quorum):
+ """
+ Same smoke-test application as test_app_upgrade, but with
dsl.store.format=HEADERS so that
+ suppress() uses the headers-aware buffer (KAFKA-20413).
+
+ The transition crosses the 4.3/trunk boundary in both directions
because the two sides write
+ the suppress changelog differently even though both tag the record as
V3:
+ - 4.3 has the dsl.store.format config but not the headers-aware
buffer, so it writes the
+ whole BufferValue into the record value.
+ - trunk writes only the plain value bytes into the record value and
ships the
+ value/timestamp/headers prefixes in extra Kafka record headers.
+
+ The suppress buffer is in-memory only, so every restart replays its
entire changelog. That
+ makes this an actual cross-format restore test: on upgrade, trunk must
restore records that
+ carry no value-part headers; on downgrade, 4.3 must cope with records
whose prefixes it never
+ learned to read.
+ """
+ old_version = SUPPRESS_HEADERS_OLD_VERSION
+ dev_version = str(DEV_VERSION)
+
+ if direction == "upgrade":
+ from_version, to_version = old_version, dev_version
+ else:
+ from_version, to_version = dev_version, old_version
+
+ self._run_app_transition(
Review Comment:
It seems no test exercises the `rolling` branch. Should we add tests for
`rolling` or just remove the dead path?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
