Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java (Java)

Code before:
            int missedCount = 0;
            for (final Map.Entry<String, Set<Integer>> entry : inputs.entrySet()) {
                missedCount += received.get(entry.getKey()).size();
            }
            System.out.println("missedRecords=" + missedCount);
        }

        VerificationResult verificationResult = pollResult.verificationResult;
        // give it one more try if it's not already passing.
        if (!verificationResult.passed()) {
            verificationResult = verifyAll(inputs, pollResult.events, true, eosEnabled);
        }
        success &= verificationResult.passed();

        System.out.println(verificationResult.result());

        System.out.println(success ? "SUCCESS" : "FAILURE");
        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) {
            return txnResult;
        }

        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");

        final long start = System.currentTimeMillis();
        try (final KafkaConsumer<String, Number> consumer = new KafkaConsumer<>(props)) {
            final PollResult pollResult = pollAndCollect(consumer, inputs, maxRecordsPerKey, eosEnabled);
            return reportAndFinalize(inputs, maxRecordsPerKey, start, eosEnabled, pollResult);
        }
    }

    private static Map<String, Set<Number>> parseRecordsForEchoTopic(
        final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events) {
        return events.containsKey("echo") ?
            events.get("echo")
                .entrySet()
                .stream()
                .map(entry -> mkEntry(
                    entry.getKey(),

Review comment (suggestion):
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.

Code after:
            for (final Map.Entry<String, Set<Integer>> entry : inputs.entrySet()) {
                missedCount += received.get(entry.getKey()).size();
            }
            System.out.println("missedRecords=" + missedCount);
        }

        VerificationResult verificationResult = pollResult.verificationResult;
        // give it one more try if it's not already passing.
        if (!verificationResult.passed()) {
            verificationResult = verifyAll(inputs, pollResult.events, true, eosEnabled);
        }
        success &= verificationResult.passed();

        System.out.println(verificationResult.result());

        System.out.println(success ? "SUCCESS" : "FAILURE");
        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.passed()) {
            return txnResult;
        }

        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");

        final long start = System.currentTimeMillis();
        try (final KafkaConsumer<String, Number> consumer = new KafkaConsumer<>(props)) {
            final PollResult pollResult = pollAndCollect(consumer, inputs, maxRecordsPerKey, eosEnabled);
            return reportAndFinalize(inputs, maxRecordsPerKey, start, eosEnabled, pollResult);
        }
    }

    private static Map<String, Set<Number>> parseRecordsForEchoTopic(
        final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events) {
        return events.containsKey("echo") ?
            events.get("echo")
                .entrySet()
                .stream()
                .map(entry -> mkEntry(
                    entry.getKey(),
                    entry.getValue().stream().map(ConsumerRecord::value).collect(Collectors.toSet()))