lianetm commented on code in PR #15202:
URL: https://github.com/apache/kafka/pull/15202#discussion_r1463501761
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -272,7 +270,65 @@ public void
testSyncAutocommitRetriedAfterRetriableException(Errors error) {
// We expect that request should have been retried on this sync commit.
assertExceptionHandling(commitRequestManger, error, true);
- assertCoordinatorDisconnect(error);
+ }
+
+ @Test
+ public void testCommitSyncThrowsCommitFailedExceptionOnFencedInstanceId() {
+
testCommitSyncFailsWithCommitFailedExceptionOnError(Errors.FENCED_INSTANCE_ID);
+ }
+
+ @Test
+ public void testCommitSyncThrowsCommitFailedExceptionOnUnknownMemberId() {
+
testCommitSyncFailsWithCommitFailedExceptionOnError(Errors.UNKNOWN_MEMBER_ID);
+ }
+
+ private void testCommitSyncFailsWithCommitFailedExceptionOnError(Errors
commitError) {
+ CommitRequestManager commitRequestManger = create(false, 100);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets =
Collections.singletonMap(
+ new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Send sync offset commit request that fails with an error that is
expected to propagate
+ // a CommitFailedException
+ Long expirationTimeMs = time.milliseconds() + retryBackoffMs;
+ CompletableFuture<Void> commitResult =
commitRequestManger.addOffsetCommitRequest(offsets,
Optional.of(expirationTimeMs), false);
+ completeOffsetCommitRequestWithError(commitRequestManger, commitError);
+ assertFutureThrows(commitResult, CommitFailedException.class);
+ }
+
+ @Test
+ public void testCommitSyncThrowsOffsetMetadataTooLargeException() {
+ // Error with metadata provided by the user should propagate the
exception, so they can handle it.
+
testCommitSyncFailsWithErrorException(Errors.OFFSET_METADATA_TOO_LARGE);
+ }
+
+ @Test
+ public void testCommitSyncThrowsInvalidCommitOffsetSizeException() {
+ // Error with data provided by the user should propagate the
exception, so they can handle it.
+
testCommitSyncFailsWithErrorException(Errors.INVALID_COMMIT_OFFSET_SIZE);
+ }
+
+ @Test
+ public void testCommitSyncThrowsGroupAuthorizationException() {
+
testCommitSyncFailsWithErrorException(Errors.GROUP_AUTHORIZATION_FAILED);
+ }
+
+ private void testCommitSyncFailsWithErrorException(Errors commitError) {
Review Comment:
Agree, missed that. All done and it did simplified a lot.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
