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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
