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

KAFKA-5556; Fix IllegalStateException in KafkaConsumer.commitSync due to missing future completion check 

This PR makes `commitOffsetsSync` method check whether future is completed after client's poll or not.

Author: umesh chaudhary <umesh9794@gmail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>

Closes #3489 from umesh9794/KAFKA-5556

(cherry picked from commit e0099e1f58)
Signed-off-by: Jason Gustafson <jason@confluent.io>
This commit is contained in:
umesh chaudhary 2017-07-20 08:54:21 -07:00 committed by Jason Gustafson
parent 2452201c57
commit 4b968afef2
2 changed files with 8 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
View File

@ -610,7 +610,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
return true;
}
if (!future.isRetriable())
if (future.failed() && !future.isRetriable())
throw future.exception();
time.sleep(retryBackoffMs);

7
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
View File

@ -1287,6 +1287,13 @@ public class ConsumerCoordinatorTest {
assertTrue(mockOffsetCommitCallback.exception instanceof IllegalArgumentException);
}
@Test
public void testCommitOffsetSyncWithoutFutureGetsCompleted() {
client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
coordinator.ensureCoordinatorReady();
assertFalse(coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), 0));
}
@Test
public void testRefreshOffset() {
client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));