root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Move pollForRecords helper method in KafkaConsumerTest 

Relocated the pollForRecords() helper method from its previous position to after the testPollIdleRatio method for improved code organization in KafkaConsumerTest.
This commit is contained in:
Kirk True 2025-09-23 17:27:09 -07:00
parent 8b33f081ce
commit 2aaca8db9d
1 changed files with 14 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
View File

@ -2720,22 +2720,6 @@ public class KafkaConsumerTest {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
@SuppressWarnings("unchecked")
private ConsumerRecords<String, String> pollForRecords() {
Timer timer = time.timer(15000);
while (timer.notExpired()) {
ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(1000));
if (!records.isEmpty())
return records;
}
throw new org.apache.kafka.common.errors.TimeoutException("no records to return");
}
@ParameterizedTest
@EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
@ -3816,6 +3800,20 @@ public void testPollIdleRatio(GroupProtocol groupProtocol) {
return new MetricName(NAME, "plugins", DESCRIPTION, expectedTags);
}
@SuppressWarnings("unchecked")
private ConsumerRecords<String, String> pollForRecords() {
Timer timer = time.timer(15000);
while (timer.notExpired()) {
ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(1000));
if (!records.isEmpty())
return records;
}
throw new org.apache.kafka.common.errors.TimeoutException("no records to return");
}
private void waitForConsumerPoll(Supplier<Boolean> testCondition, String conditionDetails) {
try {
TestUtils.waitForCondition(