Repository: apache/kafka
File: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java (Java)

Code before:

            // Assign a tolerance for how much time is allowed to pass between Consumer.poll() calls given that there
            // should be *at least* one record to read every second.
            var pollDelayTolerance = 2000;

            try (Consumer<byte[], byte[]> consumer = cluster.consumer(consumerConfig)) {
                consumer.subscribe(List.of(testTopic));

                // This is here to allow the consumer time to settle the group membership/assignment.
                awaitNonEmptyRecords(consumer, new TopicPartition(testTopic, 0));

                // Keep track of the last time the poll is invoked to ensure the deltas between invocations don't
                // exceed the delay threshold defined above.
                var beforePoll = System.currentTimeMillis();
                consumer.poll(Duration.ofSeconds(5));
                consumer.poll(Duration.ofSeconds(5));
                var afterPoll = System.currentTimeMillis();
                var pollDelay = afterPoll - beforePoll;

                if (pollDelay > pollDelayTolerance)
                    fail("Detected a stall of " + pollDelay + " ms between Consumer.poll() invocations despite a Producer producing records every " + produceDelay + " ms");
            } finally {
                executor.shutdownNow();
                // Wait for any active tasks to terminate to ensure consumer is not closed while being used from another thread
                assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS), "Executor did not terminate");
            }
        }
    }

    private ConsumerRecords<byte[], byte[]> awaitNonEmptyRecords(
        Consumer<byte[], byte[]> consumer,
        TopicPartition tp
    ) throws Exception {
        AtomicReference<ConsumerRecords<byte[], byte[]>> result = new AtomicReference<>();

        TestUtils.waitForCondition(() -> {
            var polledRecords = consumer.poll(Duration.ofSeconds(10));
            boolean hasRecords = !polledRecords.isEmpty();
            if (hasRecords) {
                result.set(polledRecords);
            }
            return hasRecords;
        }, "Timed out waiting for non-empty records from topic " + tp.topic() + " partition " + tp.partition());

        return result.get();
    }

    public static class SerializerImpl implements Serializer<byte[]> {
        private final ByteArraySerializer serializer = new ByteArraySerializer();

        @Override

Review comment (bug):
Adding a `consumer.unsubscribe()` before completing the test should help solve the flakiness @AndrewJSchofield reported.

I tried locally and got it too (around 30 iterations), and noticed it fails on the first poll to join the group and consume. This seemed to solve the flakiness on my side, up to 100 iterations, but please double check in case there is something else.

Code after:
                produceDelay,
                TimeUnit.MILLISECONDS
            );

            Map<String, Object> consumerConfig = Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT));

            // Assign a tolerance for how much time is allowed to pass between Consumer.poll() calls given that there
            // should be *at least* one record to read every second.
            var pollDelayTolerance = 2000;

            try (Consumer<byte[], byte[]> consumer = cluster.consumer(consumerConfig)) {
                consumer.subscribe(List.of(testTopic));

                // This is here to allow the consumer time to settle the group membership/assignment.
                awaitNonEmptyRecords(consumer, new TopicPartition(testTopic, 0));

                // Keep track of the last time the poll is invoked to ensure the deltas between invocations don't
                // exceed the delay threshold defined above.
                var beforePoll = System.currentTimeMillis();
                consumer.poll(Duration.ofSeconds(5));
                consumer.poll(Duration.ofSeconds(5));
                var afterPoll = System.currentTimeMillis();
                var pollDelay = afterPoll - beforePoll;

                if (pollDelay > pollDelayTolerance)
                    fail("Detected a stall of " + pollDelay + " ms between Consumer.poll() invocations despite a Producer producing records every " + produceDelay + " ms");
            } finally {
                executor.shutdownNow();
                // Wait for any active tasks to terminate to ensure consumer is not closed while being used from another thread
                assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS), "Executor did not terminate");
            }
        }
    }

    private ConsumerRecords<byte[], byte[]> awaitNonEmptyRecords(
        Consumer<byte[], byte[]> consumer,
        TopicPartition tp
    ) throws Exception {
        AtomicReference<ConsumerRecords<byte[], byte[]>> result = new AtomicReference<>();

        TestUtils.waitForCondition(() -> {
            var polledRecords = consumer.poll(Duration.ofSeconds(10));
            boolean hasRecords = !polledRecords.isEmpty();
            if (hasRecords) {
                result.set(polledRecords);
            }
            return hasRecords;
        }, "Timed out waiting for non-empty records from topic " + tp.topic() + " partition " + tp.partition());

        return result.get();
    }