Skip to content

Commit 08ef22d

Browse files
peterxclipeterxcli
authored
KAFKA-18173 Remove duplicate assertFutureError (apache#18296)
Reviewers: Chia-Ping Tsai <[email protected]>
1 parent d1aa370 commit 08ef22d

File tree

9 files changed

+161
-149
lines changed

9 files changed

+161
-149
lines changed

‎clients/src/test/java/org/apache/kafka/clients/admin/DeleteConsumerGroupOffsetsResultTest.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ public void testTopLevelErrorConstructor() throws InterruptedException {
6363
partitionFutures.completeExceptionally(Errors.GROUP_AUTHORIZATION_FAILED.exception());
6464
DeleteConsumerGroupOffsetsResult topLevelErrorResult =
6565
new DeleteConsumerGroupOffsetsResult(partitionFutures, partitions);
66-
TestUtils.assertFutureError(topLevelErrorResult.all(), GroupAuthorizationException.class);
66+
TestUtils.assertFutureThrows(topLevelErrorResult.all(), GroupAuthorizationException.class);
6767
}
6868

6969
@Test
@@ -79,9 +79,9 @@ public void testPartitionMissingInResponseErrorConstructor() throws InterruptedE
7979
DeleteConsumerGroupOffsetsResult missingPartitionResult =
8080
new DeleteConsumerGroupOffsetsResult(partitionFutures, partitions);
8181

82-
TestUtils.assertFutureError(missingPartitionResult.all(), IllegalArgumentException.class);
82+
TestUtils.assertFutureThrows(missingPartitionResult.all(), IllegalArgumentException.class);
8383
assertNull(missingPartitionResult.partitionResult(tpZero).get());
84-
TestUtils.assertFutureError(missingPartitionResult.partitionResult(tpOne), IllegalArgumentException.class);
84+
TestUtils.assertFutureThrows(missingPartitionResult.partitionResult(tpOne), IllegalArgumentException.class);
8585
}
8686

8787
@Test
@@ -110,9 +110,9 @@ private DeleteConsumerGroupOffsetsResult createAndVerifyPartitionLevelError() th
110110
DeleteConsumerGroupOffsetsResult partitionLevelErrorResult =
111111
new DeleteConsumerGroupOffsetsResult(partitionFutures, partitions);
112112

113-
TestUtils.assertFutureError(partitionLevelErrorResult.all(), UnknownTopicOrPartitionException.class);
113+
TestUtils.assertFutureThrows(partitionLevelErrorResult.all(), UnknownTopicOrPartitionException.class);
114114
assertNull(partitionLevelErrorResult.partitionResult(tpZero).get());
115-
TestUtils.assertFutureError(partitionLevelErrorResult.partitionResult(tpOne), UnknownTopicOrPartitionException.class);
115+
TestUtils.assertFutureThrows(partitionLevelErrorResult.partitionResult(tpOne), UnknownTopicOrPartitionException.class);
116116
return partitionLevelErrorResult;
117117
}
118118
}

‎clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java‎

Lines changed: 81 additions & 77 deletions
Large diffs are not rendered by default.

‎clients/src/test/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResultTest.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ public void testTopLevelErrorConstructor() throws InterruptedException {
6262
memberFutures.completeExceptionally(Errors.GROUP_AUTHORIZATION_FAILED.exception());
6363
RemoveMembersFromConsumerGroupResult topLevelErrorResult =
6464
new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove);
65-
TestUtils.assertFutureError(topLevelErrorResult.all(), GroupAuthorizationException.class);
65+
TestUtils.assertFutureThrows(topLevelErrorResult.all(), GroupAuthorizationException.class);
6666
}
6767

6868
@Test
@@ -78,9 +78,9 @@ public void testMemberMissingErrorInRequestConstructor() throws InterruptedExcep
7878
RemoveMembersFromConsumerGroupResult missingMemberResult =
7979
new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove);
8080

81-
TestUtils.assertFutureError(missingMemberResult.all(), IllegalArgumentException.class);
81+
TestUtils.assertFutureThrows(missingMemberResult.all(), IllegalArgumentException.class);
8282
assertNull(missingMemberResult.memberResult(instanceOne).get());
83-
TestUtils.assertFutureError(missingMemberResult.memberResult(instanceTwo), IllegalArgumentException.class);
83+
TestUtils.assertFutureThrows(missingMemberResult.memberResult(instanceTwo), IllegalArgumentException.class);
8484
}
8585

8686
@Test
@@ -111,9 +111,9 @@ private RemoveMembersFromConsumerGroupResult createAndVerifyMemberLevelError() t
111111
RemoveMembersFromConsumerGroupResult memberLevelErrorResult =
112112
new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove);
113113

114-
TestUtils.assertFutureError(memberLevelErrorResult.all(), FencedInstanceIdException.class);
114+
TestUtils.assertFutureThrows(memberLevelErrorResult.all(), FencedInstanceIdException.class);
115115
assertNull(memberLevelErrorResult.memberResult(instanceOne).get());
116-
TestUtils.assertFutureError(memberLevelErrorResult.memberResult(instanceTwo), FencedInstanceIdException.class);
116+
TestUtils.assertFutureThrows(memberLevelErrorResult.memberResult(instanceTwo), FencedInstanceIdException.class);
117117
return memberLevelErrorResult;
118118
}
119119
}

‎clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java‎

Lines changed: 50 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,14 @@
2525
import org.apache.kafka.common.Node;
2626
import org.apache.kafka.common.TopicPartition;
2727
import org.apache.kafka.common.Uuid;
28+
import org.apache.kafka.common.errors.GroupAuthorizationException;
29+
import org.apache.kafka.common.errors.InvalidCommitOffsetSizeException;
30+
import org.apache.kafka.common.errors.OffsetMetadataTooLarge;
2831
import org.apache.kafka.common.errors.RetriableException;
32+
import org.apache.kafka.common.errors.StaleMemberEpochException;
2933
import org.apache.kafka.common.errors.TimeoutException;
34+
import org.apache.kafka.common.errors.TopicAuthorizationException;
35+
import org.apache.kafka.common.errors.UnknownMemberIdException;
3036
import org.apache.kafka.common.message.OffsetCommitRequestData;
3137
import org.apache.kafka.common.message.OffsetCommitResponseData;
3238
import org.apache.kafka.common.message.OffsetFetchRequestData;
@@ -778,7 +784,8 @@ public void testOffsetFetchRequestErroredRequests(final Errors error) {
778784

779785
@ParameterizedTest
780786
@MethodSource("offsetFetchExceptionSupplier")
781-
public void testOffsetFetchRequestTimeoutRequests(final Errors error) {
787+
public void testOffsetFetchRequestTimeoutRequests(final Errors error,
788+
final Class<? extends Exception> expectedExceptionClass) {
782789
CommitRequestManager commitRequestManager = create(true, 100);
783790
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
784791

@@ -799,10 +806,10 @@ public void testOffsetFetchRequestTimeoutRequests(final Errors error) {
799806
assertFalse(commitRequestManager.pendingRequests.unsentOffsetFetches.isEmpty());
800807
NetworkClientDelegate.PollResult poll = commitRequestManager.poll(time.milliseconds());
801808
mimicResponse(error, poll);
802-
futures.forEach(f -> assertFutureThrows(f, TimeoutException.class));
809+
futures.forEach(f -> assertFutureThrows(f, expectedExceptionClass));
803810
assertTrue(commitRequestManager.pendingRequests.unsentOffsetFetches.isEmpty());
804811
} else {
805-
futures.forEach(f -> assertFutureThrows(f, KafkaException.class));
812+
futures.forEach(f -> assertFutureThrows(f, expectedExceptionClass));
806813
assertEmptyPendingRequests(commitRequestManager);
807814
}
808815
}
@@ -965,7 +972,9 @@ public void testOffsetCommitSyncTimeoutNotReturnedOnPollAndFails() {
965972
*/
966973
@ParameterizedTest
967974
@MethodSource("offsetCommitExceptionSupplier")
968-
public void testOffsetCommitSyncFailedWithRetriableThrowsTimeoutWhenRetryTimeExpires(final Errors error) {
975+
public void testOffsetCommitSyncFailedWithRetriableThrowsTimeoutWhenRetryTimeExpires(
976+
final Errors error,
977+
final Class<? extends Exception> expectedExceptionClass) {
969978
CommitRequestManager commitRequestManager = create(false, 100);
970979
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
971980

@@ -985,10 +994,7 @@ public void testOffsetCommitSyncFailedWithRetriableThrowsTimeoutWhenRetryTimeExp
985994
assertEquals(0, res.unsentRequests.size());
986995
assertTrue(commitResult.isDone());
987996

988-
if (error.exception() instanceof RetriableException)
989-
assertFutureThrows(commitResult, TimeoutException.class);
990-
else
991-
assertFutureThrows(commitResult, KafkaException.class);
997+
assertFutureThrows(commitResult, expectedExceptionClass);
992998
}
993999

9941000
/**
@@ -1383,40 +1389,51 @@ private void testNonRetriable(final List<CompletableFuture<Map<TopicPartition, O
13831389
*/
13841390
private static Stream<Arguments> offsetCommitExceptionSupplier() {
13851391
return Stream.of(
1386-
Arguments.of(Errors.NOT_COORDINATOR),
1387-
Arguments.of(Errors.COORDINATOR_LOAD_IN_PROGRESS),
1388-
Arguments.of(Errors.UNKNOWN_SERVER_ERROR),
1389-
Arguments.of(Errors.GROUP_AUTHORIZATION_FAILED),
1390-
Arguments.of(Errors.OFFSET_METADATA_TOO_LARGE),
1391-
Arguments.of(Errors.INVALID_COMMIT_OFFSET_SIZE),
1392-
Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION),
1393-
Arguments.of(Errors.COORDINATOR_NOT_AVAILABLE),
1394-
Arguments.of(Errors.REQUEST_TIMED_OUT),
1395-
Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED),
1396-
Arguments.of(Errors.STALE_MEMBER_EPOCH),
1397-
Arguments.of(Errors.UNKNOWN_MEMBER_ID));
1392+
// Retriable errors should result in TimeoutException when retry time expires
1393+
Arguments.of(Errors.NOT_COORDINATOR, TimeoutException.class),
1394+
Arguments.of(Errors.COORDINATOR_LOAD_IN_PROGRESS, TimeoutException.class),
1395+
Arguments.of(Errors.COORDINATOR_NOT_AVAILABLE, TimeoutException.class),
1396+
Arguments.of(Errors.REQUEST_TIMED_OUT, TimeoutException.class),
1397+
Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION, TimeoutException.class),
1398+
1399+
// Non-retriable errors should result in their specific exceptions
1400+
Arguments.of(Errors.GROUP_AUTHORIZATION_FAILED, GroupAuthorizationException.class),
1401+
Arguments.of(Errors.OFFSET_METADATA_TOO_LARGE, OffsetMetadataTooLarge.class),
1402+
Arguments.of(Errors.INVALID_COMMIT_OFFSET_SIZE, InvalidCommitOffsetSizeException.class),
1403+
Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED, TopicAuthorizationException.class),
1404+
Arguments.of(Errors.UNKNOWN_MEMBER_ID, CommitFailedException.class),
1405+
Arguments.of(Errors.STALE_MEMBER_EPOCH, CommitFailedException.class),
1406+
1407+
// Generic errors should result in KafkaException
1408+
Arguments.of(Errors.UNKNOWN_SERVER_ERROR, KafkaException.class));
13981409
}
13991410

14001411
/**
14011412
* @return {@link Errors} that could be received in {@link ApiKeys#OFFSET_FETCH} responses.
14021413
*/
14031414
private static Stream<Arguments> offsetFetchExceptionSupplier() {
14041415
return Stream.of(
1405-
Arguments.of(Errors.NOT_COORDINATOR),
1406-
Arguments.of(Errors.COORDINATOR_LOAD_IN_PROGRESS),
1407-
Arguments.of(Errors.UNKNOWN_SERVER_ERROR),
1408-
Arguments.of(Errors.GROUP_AUTHORIZATION_FAILED),
1409-
Arguments.of(Errors.OFFSET_METADATA_TOO_LARGE),
1410-
Arguments.of(Errors.INVALID_COMMIT_OFFSET_SIZE),
1411-
Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION),
1412-
Arguments.of(Errors.COORDINATOR_NOT_AVAILABLE),
1413-
Arguments.of(Errors.REQUEST_TIMED_OUT),
1414-
Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED),
1415-
Arguments.of(Errors.UNKNOWN_MEMBER_ID),
1416+
// Retriable errors should result in TimeoutException when retry time expires
1417+
Arguments.of(Errors.NOT_COORDINATOR, TimeoutException.class),
1418+
Arguments.of(Errors.COORDINATOR_LOAD_IN_PROGRESS, TimeoutException.class),
1419+
Arguments.of(Errors.COORDINATOR_NOT_AVAILABLE, TimeoutException.class),
1420+
Arguments.of(Errors.REQUEST_TIMED_OUT, TimeoutException.class),
1421+
Arguments.of(Errors.UNSTABLE_OFFSET_COMMIT, TimeoutException.class),
1422+
Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION, TimeoutException.class),
1423+
1424+
// Non-retriable errors should result in their specific exceptions
1425+
Arguments.of(Errors.GROUP_AUTHORIZATION_FAILED, GroupAuthorizationException.class),
1426+
Arguments.of(Errors.OFFSET_METADATA_TOO_LARGE, KafkaException.class),
1427+
Arguments.of(Errors.INVALID_COMMIT_OFFSET_SIZE, KafkaException.class),
1428+
1429+
Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED, KafkaException.class),
1430+
Arguments.of(Errors.UNKNOWN_MEMBER_ID, UnknownMemberIdException.class),
14161431
// Adding STALE_MEMBER_EPOCH as non-retriable here because it is only retried if a new
14171432
// member epoch is received. Tested separately.
1418-
Arguments.of(Errors.STALE_MEMBER_EPOCH),
1419-
Arguments.of(Errors.UNSTABLE_OFFSET_COMMIT));
1433+
Arguments.of(Errors.STALE_MEMBER_EPOCH, StaleMemberEpochException.class),
1434+
1435+
// Generic errors should result in KafkaException
1436+
Arguments.of(Errors.UNKNOWN_SERVER_ERROR, KafkaException.class));
14201437
}
14211438

14221439
/**

‎clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1456,7 +1456,7 @@ public void testCommitTransactionWithRecordTooLargeException() throws Exception
14561456

14571457
client.prepareResponse(endTxnResponse(Errors.NONE));
14581458
producer.beginTransaction();
1459-
TestUtils.assertFutureError(producer.send(largeRecord), RecordTooLargeException.class);
1459+
TestUtils.assertFutureThrows(producer.send(largeRecord), RecordTooLargeException.class);
14601460
assertThrows(KafkaException.class, producer::commitTransaction);
14611461
}
14621462
}
@@ -1493,7 +1493,7 @@ public void testCommitTransactionWithMetadataTimeoutForMissingTopic() throws Exc
14931493
producer.initTransactions();
14941494
producer.beginTransaction();
14951495

1496-
TestUtils.assertFutureError(producer.send(record), TimeoutException.class);
1496+
TestUtils.assertFutureThrows(producer.send(record), TimeoutException.class);
14971497
assertThrows(KafkaException.class, producer::commitTransaction);
14981498
}
14991499
}
@@ -1530,7 +1530,7 @@ public void testCommitTransactionWithMetadataTimeoutForPartitionOutOfRange() thr
15301530
producer.initTransactions();
15311531
producer.beginTransaction();
15321532

1533-
TestUtils.assertFutureError(producer.send(record), TimeoutException.class);
1533+
TestUtils.assertFutureThrows(producer.send(record), TimeoutException.class);
15341534
assertThrows(KafkaException.class, producer::commitTransaction);
15351535
}
15361536
}
@@ -1569,7 +1569,7 @@ public void testCommitTransactionWithSendToInvalidTopic() throws Exception {
15691569
producer.initTransactions();
15701570
producer.beginTransaction();
15711571

1572-
TestUtils.assertFutureError(producer.send(record), InvalidTopicException.class);
1572+
TestUtils.assertFutureThrows(producer.send(record), InvalidTopicException.class);
15731573
assertThrows(KafkaException.class, producer::commitTransaction);
15741574
}
15751575
}
@@ -2008,7 +2008,7 @@ public void testSendToInvalidTopic() throws Exception {
20082008

20092009
assertEquals(Collections.singleton(invalidTopicName),
20102010
metadata.fetch().invalidTopics(), "Cluster has incorrect invalid topic list.");
2011-
TestUtils.assertFutureError(future, InvalidTopicException.class);
2011+
TestUtils.assertFutureThrows(future, InvalidTopicException.class);
20122012

20132013
producer.close(Duration.ofMillis(0));
20142014
}

‎clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2492,15 +2492,17 @@ public void testRecordErrorPropagatedToApplication() throws InterruptedException
24922492
FutureRecordMetadata future = futureEntry.getValue();
24932493
assertTrue(future.isDone());
24942494

2495-
KafkaException exception = TestUtils.assertFutureThrows(future, KafkaException.class);
24962495
Integer index = futureEntry.getKey();
24972496
if (index == 0 || index == 2) {
2497+
InvalidRecordException exception = TestUtils.assertFutureThrows(future, InvalidRecordException.class);
24982498
assertInstanceOf(InvalidRecordException.class, exception);
24992499
assertEquals(index.toString(), exception.getMessage());
25002500
} else if (index == 3) {
2501+
InvalidRecordException exception = TestUtils.assertFutureThrows(future, InvalidRecordException.class);
25012502
assertInstanceOf(InvalidRecordException.class, exception);
25022503
assertEquals(Errors.INVALID_RECORD.message(), exception.getMessage());
25032504
} else {
2505+
KafkaException exception = TestUtils.assertFutureThrows(future, KafkaException.class);
25042506
assertEquals(KafkaException.class, exception.getClass());
25052507
}
25062508
}

‎clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.apache.kafka.common.errors.TimeoutException;
3737
import org.apache.kafka.common.errors.TopicAuthorizationException;
3838
import org.apache.kafka.common.errors.TransactionAbortableException;
39+
import org.apache.kafka.common.errors.TransactionAbortedException;
3940
import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
4041
import org.apache.kafka.common.errors.UnsupportedForMessageFormatException;
4142
import org.apache.kafka.common.errors.UnsupportedVersionException;
@@ -1529,8 +1530,8 @@ public void testTopicAuthorizationFailureInAddPartitions() throws InterruptedExc
15291530
assertAbortableError(TopicAuthorizationException.class);
15301531
sender.runOnce();
15311532

1532-
TestUtils.assertFutureThrows(firstPartitionAppend, KafkaException.class);
1533-
TestUtils.assertFutureThrows(secondPartitionAppend, KafkaException.class);
1533+
TestUtils.assertFutureThrows(firstPartitionAppend, TransactionAbortedException.class);
1534+
TestUtils.assertFutureThrows(secondPartitionAppend, TransactionAbortedException.class);
15341535
}
15351536

15361537
@Test
@@ -1577,8 +1578,8 @@ public void testCommitWithTopicAuthorizationFailureInAddPartitionsInFlight() thr
15771578
// the pending transaction commit.
15781579
sender.runOnce();
15791580
assertTrue(commitResult.isCompleted());
1580-
TestUtils.assertFutureThrows(firstPartitionAppend, KafkaException.class);
1581-
TestUtils.assertFutureThrows(secondPartitionAppend, KafkaException.class);
1581+
TestUtils.assertFutureThrows(firstPartitionAppend, TopicAuthorizationException.class);
1582+
TestUtils.assertFutureThrows(secondPartitionAppend, TopicAuthorizationException.class);
15821583
assertInstanceOf(TopicAuthorizationException.class, commitResult.error());
15831584
}
15841585

@@ -2375,7 +2376,7 @@ public void testCancelUnsentAddPartitionsAndProduceOnAbort() throws InterruptedE
23752376
assertTrue(abortResult.isSuccessful());
23762377
assertTrue(transactionManager.isReady()); // make sure we are ready for a transaction now.
23772378

2378-
TestUtils.assertFutureThrows(responseFuture, KafkaException.class);
2379+
TestUtils.assertFutureThrows(responseFuture, TransactionAbortedException.class);
23792380
}
23802381

23812382
@Test
@@ -2401,7 +2402,7 @@ public void testAbortResendsAddPartitionErrorIfRetried() throws InterruptedExcep
24012402
assertTrue(abortResult.isSuccessful());
24022403
assertTrue(transactionManager.isReady()); // make sure we are ready for a transaction now.
24032404

2404-
TestUtils.assertFutureThrows(responseFuture, KafkaException.class);
2405+
TestUtils.assertFutureThrows(responseFuture, TransactionAbortedException.class);
24052406
}
24062407

24072408
@Test

0 commit comments

Comments
 (0)