Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,12 @@ AbstractRequest.Builder<?> requestBuilder() {

@Override
public String toString() {
return "UnsentRequest(builder=" + requestBuilder + ")";
return "UnsentRequest{" +
"requestBuilder=" + requestBuilder +
", handler=" + handler +
", node=" + node +
", timer=" + timer +
'}';
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -333,7 +333,7 @@ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition
final OffsetFetchApplicationEvent event = new OffsetFetchApplicationEvent(partitions);
eventHandler.add(event);
try {
return event.complete(Duration.ofMillis(100));
return event.complete(timeout);
} catch (InterruptedException e) {
throw new InterruptException(e);
} catch (TimeoutException e) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,10 @@
package kafka.api

import kafka.utils.TestUtils.waitUntilTrue
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test

import scala.jdk.CollectionConverters._

class BaseAsyncConsumerTest extends AbstractConsumerTest {
@Test
Expand All @@ -29,10 +31,12 @@ class BaseAsyncConsumerTest extends AbstractConsumerTest {
val startingTimestamp = System.currentTimeMillis()
val cb = new CountConsumerCommitCallback
sendRecords(producer, numRecords, tp, startingTimestamp = startingTimestamp)
consumer.assign(List(tp).asJava)
consumer.commitAsync(cb)
waitUntilTrue(() => {
cb.successCount == 1
}, "wait until commit is completed successfully", 5000)
assertTrue(consumer.assignment.contains(tp))
}

@Test
Expand All @@ -42,6 +46,9 @@ class BaseAsyncConsumerTest extends AbstractConsumerTest {
val numRecords = 10000
val startingTimestamp = System.currentTimeMillis()
sendRecords(producer, numRecords, tp, startingTimestamp = startingTimestamp)
consumer.assign(List(tp).asJava)
consumer.commitSync();

assertTrue(consumer.assignment.contains(tp))
}
}