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

Code before:
    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;
        }
    }

    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;
    }

    private static PollResult pollAndCollect(
            final KafkaConsumer<String, Number> consumer,
            final Map<String, Set<Integer>> inputs,
            final int maxRecordsPerKey,
            final boolean eosEnabled) {
        final int recordsGenerated = inputs.size() * maxRecordsPerKey;
        final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events = new HashMap<>();
        VerificationResult verificationResult = new VerificationResult(false, "no results yet");
        final long start = System.currentTimeMillis();
        int recordsProcessed = 0;

        final List<TopicPartition> partitions = getAllPartitions(consumer, NUMERIC_VALUE_TOPICS);
        consumer.assign(partitions);
        consumer.seekToBeginning(partitions);
        final Map<String, AtomicInteger> processed =
            Stream.of(NUMERIC_VALUE_TOPICS)
                  .collect(Collectors.toMap(t -> t, t -> new AtomicInteger(0)));

        int retry = 0;
        while (System.currentTimeMillis() - start < TimeUnit.MINUTES.toMillis(6)) {
            final ConsumerRecords<String, Number> records = consumer.poll(Duration.ofSeconds(5));
            if (records.isEmpty() && recordsProcessed >= recordsGenerated) {
                verificationResult = verifyAll(inputs, events, false, eosEnabled);

Review comment (suggestion):
As above -- why do we convert a "passed" into `null` ? Kinda confusing.

Code after:
    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;
        }
    }

    private static VerificationResult preVerifyTransactions(final String kafka, final boolean eosEnabled) {
        if (!eosEnabled) {
            return new VerificationResult(true, "EOS is disabled; skipping transaction verification");
        }

        final VerificationResult txnResult = verifyAllTransactionFinished(kafka);
        if (!txnResult.passed()) {
            System.err.println("Transaction verification failed: " + txnResult.result());
            System.out.println("FAILED");
        }
        return txnResult;
    }

    private static PollResult pollAndCollect(
            final KafkaConsumer<String, Number> consumer,
            final Map<String, Set<Integer>> inputs,
            final int maxRecordsPerKey,
            final boolean eosEnabled) {
        final int recordsGenerated = inputs.size() * maxRecordsPerKey;
        final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events = new HashMap<>();
        VerificationResult verificationResult = new VerificationResult(false, "no results yet");
        final long start = System.currentTimeMillis();
        int recordsProcessed = 0;

        final List<TopicPartition> partitions = getAllPartitions(consumer, NUMERIC_VALUE_TOPICS);
        consumer.assign(partitions);
        consumer.seekToBeginning(partitions);
        final Map<String, AtomicInteger> processed =
            Stream.of(NUMERIC_VALUE_TOPICS)
                  .collect(Collectors.toMap(t -> t, t -> new AtomicInteger(0)));

        int retry = 0;
        while (System.currentTimeMillis() - start < TimeUnit.MINUTES.toMillis(6)) {
            final ConsumerRecords<String, Number> records = consumer.poll(Duration.ofSeconds(5));
            if (records.isEmpty() && recordsProcessed >= recordsGenerated) {
                verificationResult = verifyAll(inputs, events, false, eosEnabled);
                if (verificationResult.passed()) {