You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "RaidenE1 (via GitHub)" <gi...@apache.org> on 2025/10/16 21:21:45 UTC
[PR] MINOR: Migrate `verificationAllTransactionComplete` from EOS system test to smoke test [kafka]
RaidenE1 opened a new pull request, #20718:
URL: https://github.com/apache/kafka/pull/20718
First step of [KAFKA-8080](https://issues.apache.org/jira/projects/KAFKA/issues/KAFKA-8080?filter=myopenissues)
1. migrate `verificationAllTransactionComplete`
2. add `VerificationResult` to smoke test utils and make `verificationAllTransactionComplete` return `VerificationResult` to avoid extra exception handling
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Migrate `verificationAllTransactionComplete` from EOS system test to smoke test [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #20718:
URL: https://github.com/apache/kafka/pull/20718#issuecomment-3573165163
Thanks for the PR. Merged to `trunk`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Migrate `verificationAllTransactionComplete` from EOS system test to smoke test [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20718:
URL: https://github.com/apache/kafka/pull/20718#discussion_r2496804395
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -383,60 +386,79 @@ public static VerificationResult verify(final String kafka,
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, NumberDeserializer.class);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
- final KafkaConsumer<String, Number> consumer = new KafkaConsumer<>(props);
- final List<TopicPartition> partitions = getAllPartitions(consumer, NUMERIC_VALUE_TOPICS);
- consumer.assign(partitions);
- consumer.seekToBeginning(partitions);
+ // Verify all transactions are finished before proceeding with data verification
+ if (eosEnabled) {
+ final Properties txnProps = new Properties();
+ txnProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "verifier");
+ txnProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
+ txnProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+ txnProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+ txnProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString());
+
+ final VerificationResult txnResult;
+ try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(txnProps)) {
Review Comment:
Seem we are creating a consumer here, and also below. Could we re-use the same consumer? Ie use a top level `try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(...))` and nest `if (eosEnabled)` block inside this try-catch, as well as the other verification we do later?
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -383,60 +386,79 @@ public static VerificationResult verify(final String kafka,
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, NumberDeserializer.class);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
- final KafkaConsumer<String, Number> consumer = new KafkaConsumer<>(props);
- final List<TopicPartition> partitions = getAllPartitions(consumer, NUMERIC_VALUE_TOPICS);
- consumer.assign(partitions);
- consumer.seekToBeginning(partitions);
+ // Verify all transactions are finished before proceeding with data verification
+ if (eosEnabled) {
+ final Properties txnProps = new Properties();
Review Comment:
Seems this are the same config we setup above for `props`. Seems we could just reuse the ones from above?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Migrate `verificationAllTransactionComplete` from EOS system test to smoke test [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #20718:
URL: https://github.com/apache/kafka/pull/20718
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Migrate `verificationAllTransactionComplete` from EOS system test to smoke test [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20718:
URL: https://github.com/apache/kafka/pull/20718#issuecomment-3440728086
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Migrate `verificationAllTransactionComplete` from EOS system test to smoke test [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20718:
URL: https://github.com/apache/kafka/pull/20718#discussion_r2548677821
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -478,6 +511,29 @@ public static VerificationResult verify(final String kafka,
return verificationResult;
}
+ public static VerificationResult verify(final String kafka,
+ final Map<String, Set<Integer>> inputs,
+ final int maxRecordsPerKey,
+ final boolean eosEnabled) {
+ final VerificationResult txnResult = preVerifyTransactions(kafka, eosEnabled);
+ if (txnResult != null) {
Review Comment:
This does not seem to be easy to reason about? If `txnResult` is not null and passed, we only know that all transactions completed, but we cannot stop the verification yet -- we could only exit early if we get a "false" `VerifcationResult`.
This code does only make sense if one knows how `preVerifyTransactions` works, and it would return `null` for "passed == true" case, but it make the code hard to ready.
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -372,33 +375,53 @@ public Number deserialize(final String topic, final byte[] data) {
}
}
- public static VerificationResult verify(final String kafka,
- final Map<String, Set<Integer>> inputs,
- final int maxRecordsPerKey,
- final boolean eosEnabled) {
- final Properties props = new Properties();
- props.put(ConsumerConfig.CLIENT_ID_CONFIG, "verifier");
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, NumberDeserializer.class);
- props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+ private static class PollResult {
+ final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events;
+ final int recordsProcessed;
+ final VerificationResult verificationResult;
+
+ PollResult(final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events,
+ final int recordsProcessed,
+ final VerificationResult verificationResult) {
+ this.events = events;
+ this.recordsProcessed = recordsProcessed;
+ this.verificationResult = verificationResult;
+ }
+ }
- final KafkaConsumer<String, Number> consumer = new KafkaConsumer<>(props);
- final List<TopicPartition> partitions = getAllPartitions(consumer, NUMERIC_VALUE_TOPICS);
- consumer.assign(partitions);
- consumer.seekToBeginning(partitions);
+ private static VerificationResult preVerifyTransactions(final String kafka, final boolean eosEnabled) {
+ if (!eosEnabled) {
+ return null;
+ }
+
+ final VerificationResult txnResult = verifyAllTransactionFinished(kafka);
+ if (!txnResult.passed()) {
+ System.err.println("Transaction verification failed: " + txnResult.result());
+ System.out.println("FAILED");
+ return txnResult;
+ }
+ return null;
Review Comment:
As above -- why do we convert a "passed" into `null` ? Kinda confusing.
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -372,33 +375,53 @@ public Number deserialize(final String topic, final byte[] data) {
}
}
- public static VerificationResult verify(final String kafka,
- final Map<String, Set<Integer>> inputs,
- final int maxRecordsPerKey,
- final boolean eosEnabled) {
- final Properties props = new Properties();
- props.put(ConsumerConfig.CLIENT_ID_CONFIG, "verifier");
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, NumberDeserializer.class);
- props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+ private static class PollResult {
+ final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events;
+ final int recordsProcessed;
+ final VerificationResult verificationResult;
+
+ PollResult(final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events,
+ final int recordsProcessed,
+ final VerificationResult verificationResult) {
+ this.events = events;
+ this.recordsProcessed = recordsProcessed;
+ this.verificationResult = verificationResult;
+ }
+ }
- final KafkaConsumer<String, Number> consumer = new KafkaConsumer<>(props);
- final List<TopicPartition> partitions = getAllPartitions(consumer, NUMERIC_VALUE_TOPICS);
- consumer.assign(partitions);
- consumer.seekToBeginning(partitions);
+ private static VerificationResult preVerifyTransactions(final String kafka, final boolean eosEnabled) {
+ if (!eosEnabled) {
+ return null;
Review Comment:
Why do we return `null`? Might be easier to return a `VerificationResult` with `passed == true`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Migrate `verificationAllTransactionComplete` from EOS system test to smoke test [kafka]
Posted by "RaidenE1 (via GitHub)" <gi...@apache.org>.
RaidenE1 commented on code in PR #20718:
URL: https://github.com/apache/kafka/pull/20718#discussion_r2550774907
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -478,6 +511,29 @@ public static VerificationResult verify(final String kafka,
return verificationResult;
}
+ public static VerificationResult verify(final String kafka,
+ final Map<String, Set<Integer>> inputs,
+ final int maxRecordsPerKey,
+ final boolean eosEnabled) {
+ final VerificationResult txnResult = preVerifyTransactions(kafka, eosEnabled);
+ if (txnResult != null) {
Review Comment:
Make sense, updated
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org