ConsumerCoordinator extends AbstractCoordinator { public void commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets) { if (offsets.isEmpty()) { return; } while (true) { ensureCoordinatorReady(); RequestFuture<Void> future = sendOffsetCommitRequest(offsets); client.poll(future); if (future.succeeded()) { if (interceptors != null) { interceptors.onCommit(offsets); } return; } if (!future.isRetriable()) { throw future.exception(); } time.sleep(retryBackoffMs); } } ConsumerCoordinator(ConsumerNetworkClient client,
        String groupId,
        int sessionTimeoutMs,
        int heartbeatIntervalMs,
        List<PartitionAssignor> assignors,
        Metadata metadata,
        SubscriptionState subscriptions,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs,
        OffsetCommitCallback defaultOffsetCommitCallback,
        boolean autoCommitEnabled,
        long autoCommitIntervalMs,
        ConsumerInterceptors<?, ?> interceptors,
        boolean excludeInternalTopics); @Override String protocolType(); @Override List<ProtocolMetadata> metadata(); @Override boolean needRejoin(); void refreshCommittedOffsetsIfNeeded(); Map<TopicPartition, OffsetAndMetadata> fetchCommittedOffsets(Set<TopicPartition> partitions); void ensurePartitionAssignment(); @Override void close(); void commitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback); void commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets);  }
@Test public void testCommitOffsetSyncNotCoordinator() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NOT_COORDINATOR_FOR_GROUP.code()))); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L))); }
@Test public void testCommitOffsetSyncCoordinatorNotAvailable() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()))); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L))); }
@Test public void testCommitOffsetSyncCoordinatorDisconnected() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code())), true); client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L))); }
@Test(expected = OffsetMetadataTooLarge.class) public void testCommitOffsetMetadataTooLarge() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.OFFSET_METADATA_TOO_LARGE.code()))); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata"))); }
@Test(expected = CommitFailedException.class) public void testCommitOffsetIllegalGeneration() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.ILLEGAL_GENERATION.code()))); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata"))); }
@Test(expected = CommitFailedException.class) public void testCommitOffsetUnknownMemberId() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.UNKNOWN_MEMBER_ID.code()))); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata"))); }
@Test(expected = CommitFailedException.class) public void testCommitOffsetRebalanceInProgress() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.REBALANCE_IN_PROGRESS.code()))); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata"))); }
@Test(expected = KafkaException.class) public void testCommitOffsetSyncCallbackWithNonRetriableException() { client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.UNKNOWN.code())), false); coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L))); }