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

Code before:
                        Utils.sleep(produceWait);
                        producer.send(new ProducerRecord<>(testTopic, TestUtils.randomBytes(64))).get();
                    } catch (InterruptedException e) {
                        break;
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
            });
            producerThread.start();

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

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

                // This is just to wait until the group membership and assignment is in place.
                awaitNonEmptyRecords(consumer, new TopicPartition(testTopic, 0));

                var testTimer = Time.SYSTEM.timer(testTimeout);

                // 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 lastPoll = System.currentTimeMillis();

                while (testTimer.notExpired()) {
                    try {
                        consumer.poll(Duration.ofMillis(10000));
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }

                    var currPoll = System.currentTimeMillis();
                    var pollDelay = currPoll - lastPoll;

                    if (pollDelay > delayTolerance) {
                        fail("Detected a stall of " + pollDelay + " ms between Consumer.poll() invocations despite a Producer producing records every " + produceWait + " ms");
                        break;
                    }

                    lastPoll = currPoll;
                    testTimer.update();
                }
            } finally {
                producerThread.interrupt();
                producerThread.join();
            }
        }
    }

    private ConsumerRecords<byte[], byte[]> awaitNonEmptyRecords(

Review comment (refactor):
this means that this test will run for the full 1min 15s (testTimeout=75s), but can't we simplify and detect the issue much sooner? 

The faulty behaviour is that one poll runs fast, and the next one takes the full pollTimeout, right? So maybe I'm missing something but wouldn't 2 polls be enough? (also considering that `awaitNonEmptyRecords` does poll before the actual call to consumer.poll on ln 1670). 

Code after:
            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
        public byte[] serialize(String topic, byte[] data) {
            throw new RuntimeException("This method should not be called");
        }

        @Override