You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jy...@apache.org on 2026/09/07 12:24:45 UTC
(kafka) branch trunk updated: KAFKA-20942: Refactor empty catch blocks in clients module tests to use assertThrows (#23187)
This is an automated email from the ASF dual-hosted git repository.
m1a2st pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 5ce7e342502 KAFKA-20942: Refactor empty catch blocks in clients module tests to use assertThrows (#23187)
5ce7e342502 is described below
commit 5ce7e34250273652e5496846a5278e87e78b13fd
Author: Kapil Hadoltikar <14...@users.noreply.github.com>
AuthorDate: Mon Sep 7 17:54:35 2026 +0530
KAFKA-20942: Refactor empty catch blocks in clients module tests to use assertThrows (#23187)
* Empty catch blocks can swallow unexpected errors, bugs, or exceptions
during test execution.
* Refactored empty catch blocks in the clients module tests to use JUnit
5's assertThrows for strict and safe exception handling.
### Changes
* Refactored empty catch blocks to use JUnit 5's assertThrows across
several test areas in the clients module:
* Admin & Consumer: `DescribeUserScramCredentialsResultTest`,
consumer internals `AbstractCoordinatorTest`, `ConsumerCoordinatorTest`,
`FetcherTest`, `FetchRequestManagerTest`, and `KafkaConsumerTest`.
* Producer & Internals: `BufferPoolTest`, `ProducerBatchTest`,
`ProducerMetadataTest`, `TransactionManagerTest`, and
`KafkaProducerTest`.
* Common Config, Metrics & Network: `ConfigDefTest`, `TopicTest`,
`ApiMessageTypeTest`, `KafkaMbeanTest`, `SensorTest`, and
`SelectorTest`.
* Security & Utils: `FileRecordsTest`, `Sasl`, `OAuthbearer`, and
`ScramMessagesTest`, `SslFactoryTest`, `JaasContextTest`,
`ImplicitLinkedHashCollectionTest`, and `UtilsTest`.
### Testing
* Ran ./gradlew :clients:test locally and verified all tests passed.
* All CI checks passed successfully.
Reviewers: Ken Huang <s7...@gmail.com>
Signed-off-by: Kapil Hadoltikar <ha...@gmail.com>
---
.../DescribeUserScramCredentialsResultTest.java | 51 +++----------------
.../kafka/clients/consumer/KafkaConsumerTest.java | 9 ++--
.../internals/AbstractCoordinatorTest.java | 35 ++-----------
.../internals/ConsumerCoordinatorTest.java | 13 +----
.../internals/FetchRequestManagerTest.java | 9 +---
.../clients/consumer/internals/FetcherTest.java | 9 +---
.../consumer/internals/OffsetFetcherTest.java | 8 +--
.../internals/ShareConsumeRequestManagerTest.java | 9 +---
.../internals/TopicMetadataFetcherTest.java | 9 +---
.../kafka/clients/producer/KafkaProducerTest.java | 5 +-
.../clients/producer/internals/BufferPoolTest.java | 36 +++-----------
.../producer/internals/ProducerBatchTest.java | 7 +--
.../producer/internals/ProducerMetadataTest.java | 15 +-----
.../producer/internals/RecordAccumulatorTest.java | 9 +---
.../clients/producer/internals/SenderTest.java | 58 +++++++---------------
.../producer/internals/TransactionManagerTest.java | 8 +--
.../apache/kafka/common/config/ConfigDefTest.java | 8 +--
.../apache/kafka/common/internals/TopicTest.java | 9 +---
.../kafka/common/message/ApiMessageTypeTest.java | 8 +--
.../kafka/common/metrics/KafkaMbeanTest.java | 7 +--
.../apache/kafka/common/metrics/SensorTest.java | 8 +--
.../common/network/SaslChannelBuilderTest.java | 11 ++--
.../apache/kafka/common/network/SelectorTest.java | 7 +--
.../protocol/types/ProtocolSerializationTest.java | 20 +++++---
.../protocol/types/RawTaggedFieldWriterTest.java | 19 ++-----
.../internal/AbstractLegacyRecordBatchTest.java | 23 +++------
.../common/record/internal/FileRecordsTest.java | 8 +--
.../common/requests/LeaveGroupRequestTest.java | 29 +++++------
.../kafka/common/security/JaasContextTest.java | 18 ++-----
.../authenticator/SaslAuthenticatorTest.java | 29 ++---------
.../unsecured/OAuthBearerScopeUtilsTest.java | 9 +---
.../scram/internals/ScramMessagesTest.java | 9 +---
.../kafka/common/security/ssl/SslFactoryTest.java | 30 ++---------
.../org/apache/kafka/common/utils/UtilsTest.java | 7 +--
.../ImplicitLinkedHashCollectionTest.java | 15 +-----
35 files changed, 134 insertions(+), 430 deletions(-)
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/DescribeUserScramCredentialsResultTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/DescribeUserScramCredentialsResultTest.java
index 8953a25aac8..bcac376da28 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/DescribeUserScramCredentialsResultTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/DescribeUserScramCredentialsResultTest.java
@@ -28,7 +28,7 @@ import java.util.Collections;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assertions.assertThrows;
public class DescribeUserScramCredentialsResultTest {
@Test
@@ -36,24 +36,9 @@ public class DescribeUserScramCredentialsResultTest {
KafkaFutureImpl<DescribeUserScramCredentialsResponseData> dataFuture = new KafkaFutureImpl<>();
dataFuture.completeExceptionally(new RuntimeException());
DescribeUserScramCredentialsResult results = new DescribeUserScramCredentialsResult(dataFuture);
- try {
- results.all().get();
- fail("expected all() to fail when there is a top-level error");
- } catch (Exception expected) {
- // ignore, expected
- }
- try {
- results.users().get();
- fail("expected users() to fail when there is a top-level error");
- } catch (Exception expected) {
- // ignore, expected
- }
- try {
- results.description("whatever").get();
- fail("expected description() to fail when there is a top-level error");
- } catch (Exception expected) {
- // ignore, expected
- }
+ assertThrows(Exception.class, () -> results.all().get(), "expected all() to fail when there is a top-level error");
+ assertThrows(Exception.class, () -> results.users().get(), "expected users() to fail when there is a top-level error");
+ assertThrows(Exception.class, () -> results.description("whatever").get(), "expected description() to fail when there is a top-level error");
}
@Test
@@ -70,27 +55,12 @@ public class DescribeUserScramCredentialsResultTest {
new DescribeUserScramCredentialsResponseData.DescribeUserScramCredentialsResult().setUser(unknownUser).setErrorCode(Errors.RESOURCE_NOT_FOUND.code()),
new DescribeUserScramCredentialsResponseData.DescribeUserScramCredentialsResult().setUser(failedUser).setErrorCode(Errors.DUPLICATE_RESOURCE.code()))));
DescribeUserScramCredentialsResult results = new DescribeUserScramCredentialsResult(dataFuture);
- try {
- results.all().get();
- fail("expected all() to fail when there is a user-level error");
- } catch (Exception expected) {
- // ignore, expected
- }
+ assertThrows(Exception.class, () -> results.all().get(), "expected all() to fail when there is a user-level error");
assertEquals(Arrays.asList(goodUser, failedUser), results.users().get(), "Expected 2 users with credentials");
UserScramCredentialsDescription goodUserDescription = results.description(goodUser).get();
assertEquals(new UserScramCredentialsDescription(goodUser, Collections.singletonList(new ScramCredentialInfo(scramSha256, iterations))), goodUserDescription);
- try {
- results.description(failedUser).get();
- fail("expected description(failedUser) to fail when there is a user-level error");
- } catch (Exception expected) {
- // ignore, expected
- }
- try {
- results.description(unknownUser).get();
- fail("expected description(unknownUser) to fail when there is no such user");
- } catch (Exception expected) {
- // ignore, expected
- }
+ assertThrows(Exception.class, () -> results.description(failedUser).get(), "expected description(failedUser) to fail when there is a user-level error");
+ assertThrows(Exception.class, () -> results.description(unknownUser).get(), "expected description(unknownUser) to fail when there is no such user");
}
@Test
@@ -110,11 +80,6 @@ public class DescribeUserScramCredentialsResultTest {
UserScramCredentialsDescription goodUserDescriptionViaAll = allResults.get(goodUser);
assertEquals(new UserScramCredentialsDescription(goodUser, Collections.singletonList(new ScramCredentialInfo(scramSha256, iterations))), goodUserDescriptionViaAll);
assertEquals(goodUserDescriptionViaAll, results.description(goodUser).get(), "Expected same thing via all() and description()");
- try {
- results.description(unknownUser).get();
- fail("expected description(unknownUser) to fail when there is no such user even when all() succeeds");
- } catch (Exception expected) {
- // ignore, expected
- }
+ assertThrows(Exception.class, () -> results.description(unknownUser).get(), "expected description(unknownUser) to fail when there is no such user even when all() succeeds");
}
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index 58eb53894f2..9fd0a114fc0 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -2501,12 +2501,11 @@ public class KafkaConsumerTest {
// Close task should not complete until commit succeeds or close times out
// if close timeout is not zero.
- try {
+ if (closeTimeoutMs != 0) {
+ assertThrows(TimeoutException.class, () -> future.get(100, TimeUnit.MILLISECONDS), "Close completed without waiting for commit or leave response");
+ } else {
+ // Handle the case where timeout is 0, if needed
future.get(100, TimeUnit.MILLISECONDS);
- if (closeTimeoutMs != 0)
- fail("Close completed without waiting for commit or leave response");
- } catch (TimeoutException swallow) {
- // Expected exception
}
// Ensure close has started and queued at least one more request after commitAsync.
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
index dd37130b531..1a5e611c182 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
@@ -1427,12 +1427,7 @@ public class AbstractCoordinatorTest {
}, joinGroupFollowerResponse(1, memberId, leaderId, Errors.NONE));
mockClient.prepareResponse(syncGroupResponse(Errors.NONE));
AtomicBoolean heartbeatReceived = prepareFirstHeartbeat();
-
- try {
- coordinator.ensureActiveGroup();
- fail("Should have woken up from ensureActiveGroup()");
- } catch (WakeupException ignored) {
- }
+ assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
assertEquals(1, coordinator.onJoinPrepareInvokes);
assertEquals(0, coordinator.onJoinCompleteInvokes);
@@ -1465,12 +1460,7 @@ public class AbstractCoordinatorTest {
}, joinGroupFollowerResponse(1, memberId, leaderId, Errors.NONE));
mockClient.prepareResponse(syncGroupResponse(Errors.NONE));
AtomicBoolean heartbeatReceived = prepareFirstHeartbeat();
-
- try {
- coordinator.ensureActiveGroup();
- fail("Should have woken up from ensureActiveGroup()");
- } catch (WakeupException ignored) {
- }
+ assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
assertEquals(1, coordinator.onJoinPrepareInvokes);
assertEquals(0, coordinator.onJoinCompleteInvokes);
@@ -1500,12 +1490,7 @@ public class AbstractCoordinatorTest {
}, joinGroupFollowerResponse(1, memberId, leaderId, Errors.NONE));
mockClient.prepareResponse(syncGroupResponse(Errors.NONE));
AtomicBoolean heartbeatReceived = prepareFirstHeartbeat();
-
- try {
- coordinator.ensureActiveGroup();
- fail("Should have woken up from ensureActiveGroup()");
- } catch (WakeupException ignored) {
- }
+ assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
assertEquals(1, coordinator.onJoinPrepareInvokes);
assertEquals(0, coordinator.onJoinCompleteInvokes);
@@ -1599,12 +1584,7 @@ public class AbstractCoordinatorTest {
consumerClient.wakeup();
return isSyncGroupRequest;
}, syncGroupResponse(Errors.NONE));
-
- try {
- coordinator.ensureActiveGroup();
- fail("Should have woken up from ensureActiveGroup()");
- } catch (WakeupException ignored) {
- }
+ assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
assertEquals(1, coordinator.onJoinPrepareInvokes);
assertEquals(0, coordinator.onJoinCompleteInvokes);
@@ -1654,12 +1634,7 @@ public class AbstractCoordinatorTest {
mockClient.prepareResponse(joinGroupFollowerResponse(1, memberId, leaderId, Errors.NONE));
mockClient.prepareResponse(syncGroupResponse(Errors.NONE));
AtomicBoolean heartbeatReceived = prepareFirstHeartbeat();
-
- try {
- coordinator.ensureActiveGroup();
- fail("Should have woken up from ensureActiveGroup()");
- } catch (WakeupException ignored) {
- }
+ assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
assertEquals(1, coordinator.onJoinPrepareInvokes);
assertEquals(0, coordinator.onJoinCompleteInvokes);
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
index 6ac647665b0..5a44522df33 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -2303,11 +2303,7 @@ public abstract class ConsumerCoordinatorTest {
client.prepareResponse(syncGroupResponse(Collections.singletonList(partition), Errors.NONE));
// The first call to poll should raise the exception from the rebalance listener
- try {
- coordinator.poll(time.timer(Long.MAX_VALUE));
- fail("Expected exception thrown from assignment callback");
- } catch (WakeupException e) {
- }
+ assertThrows(WakeupException.class, () -> coordinator.poll(time.timer(Long.MAX_VALUE)), "Expected exception thrown from assignment callback");
// The second call should retry the assignment callback and succeed
coordinator.poll(time.timer(Long.MAX_VALUE));
@@ -3981,12 +3977,7 @@ public abstract class ConsumerCoordinatorTest {
Thread.sleep(200);
if (expectedMinTimeMs > 0) {
time.sleep(expectedMinTimeMs - 1);
- try {
- future.get(500, TimeUnit.MILLISECONDS);
- fail("Close completed ungracefully without waiting for timeout");
- } catch (TimeoutException e) {
- // Expected timeout
- }
+ assertThrows(TimeoutException.class, () -> future.get(500, TimeUnit.MILLISECONDS), "Close completed ungracefully without waiting for timeout");
}
if (expectedMaxTimeMs >= 0)
time.sleep(expectedMaxTimeMs - expectedMinTimeMs + 2);
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
index 37056ac19df..cd1bd015392 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
@@ -1213,13 +1213,8 @@ public class FetchRequestManagerTest {
ensureBlockOnRecord(1L);
seekAndConsumeRecord(buffer, 2L);
ensureBlockOnRecord(3L);
- try {
- // For a record that cannot be retrieved from the iterator, we cannot seek over it within the batch.
- seekAndConsumeRecord(buffer, 4L);
- fail("Should have thrown exception when fail to retrieve a record from iterator.");
- } catch (KafkaException ke) {
- // let it go
- }
+ // For a record that cannot be retrieved from the iterator, we cannot seek over it within the batch.
+ assertThrows(KafkaException.class, () -> seekAndConsumeRecord(buffer, 4L), "Should have thrown exception when fail to retrieve a record from iterator.");
ensureBlockOnRecord(4L);
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
index 4deafeac25e..2c262201300 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
@@ -960,13 +960,8 @@ public class FetcherTest {
ensureBlockOnRecord(1L);
seekAndConsumeRecord(buffer, 2L);
ensureBlockOnRecord(3L);
- try {
- // For a record that cannot be retrieved from the iterator, we cannot seek over it within the batch.
- seekAndConsumeRecord(buffer, 4L);
- fail("Should have thrown exception when fail to retrieve a record from iterator.");
- } catch (KafkaException ke) {
- // let it go
- }
+ // For a record that cannot be retrieved from the iterator, we cannot seek over it within the batch.
+ assertThrows(KafkaException.class, () -> seekAndConsumeRecord(buffer, 4L), "Should have thrown exception when fail to retrieve a record from iterator.");
ensureBlockOnRecord(4L);
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherTest.java
index afab70cad4f..8c2a352708f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherTest.java
@@ -659,12 +659,8 @@ public class OffsetFetcherTest {
consumerClient.pollNoWakeup();
assertFalse(subscriptions.hasValidPosition(tp0));
- try {
- offsetFetcher.resetPositionsIfNeeded();
- fail("Expected authorization error to be raised");
- } catch (TopicAuthorizationException e) {
- assertEquals(singleton(tp0.topic()), e.unauthorizedTopics());
- }
+ TopicAuthorizationException e = assertThrows(TopicAuthorizationException.class, () -> offsetFetcher.resetPositionsIfNeeded(), "Expected authorization error to be raised");
+ assertEquals(singleton(tp0.topic()), e.unauthorizedTopics());
// The exception should clear after being raised, but no retry until the backoff
offsetFetcher.resetPositionsIfNeeded();
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
index b36d5f02ab8..f156ba4f369 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
@@ -125,7 +125,6 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
@@ -1964,12 +1963,8 @@ public class ShareConsumeRequestManagerTest {
assertEquals(1, sendFetches());
client.prepareResponse(fullFetchResponse(tip0, records, emptyAcquiredRecords, Errors.TOPIC_AUTHORIZATION_FAILED));
networkClientDelegate.poll(time.timer(0));
- try {
- collectFetch();
- fail("collectFetch should have thrown a TopicAuthorizationException");
- } catch (TopicAuthorizationException e) {
- assertEquals(Set.of(topicName), e.unauthorizedTopics());
- }
+ TopicAuthorizationException e = assertThrows(TopicAuthorizationException.class, () -> collectFetch(), "collectFetch should have thrown a TopicAuthorizationException");
+ assertEquals(Set.of(topicName), e.unauthorizedTopics());
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcherTest.java
index 997ea869f21..736edfb221f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcherTest.java
@@ -52,7 +52,6 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.fail;
public class TopicMetadataFetcherTest {
@@ -132,12 +131,8 @@ public class TopicMetadataFetcherTest {
buildFetcher();
assignFromUser(singleton(tp0));
client.prepareResponse(newMetadataResponse(Errors.TOPIC_AUTHORIZATION_FAILED));
- try {
- topicMetadataFetcher.getAllTopicMetadata(time.timer(10L));
- fail();
- } catch (TopicAuthorizationException e) {
- assertEquals(singleton(topicName), e.unauthorizedTopics());
- }
+ TopicAuthorizationException e = assertThrows(TopicAuthorizationException.class, () -> topicMetadataFetcher.getAllTopicMetadata(time.timer(50L)));
+ assertEquals(singleton(topicName), e.unauthorizedTopics());
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
index a47fbcf5846..bac14a59e6a 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
@@ -727,10 +727,7 @@ public class KafkaProducerTest {
});
// Close producer should not complete until send succeeds
- try {
- future.get(100, TimeUnit.MILLISECONDS);
- fail("Close completed without waiting for send");
- } catch (java.util.concurrent.TimeoutException expected) { /* ignore */ }
+ assertThrows(java.util.concurrent.TimeoutException.class, () -> future.get(100, TimeUnit.MILLISECONDS), "Close completed without waiting for send");
// Ensure send has started
client.waitForRequests(1, 1000);
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java
index 727368e8edd..3bb4bf53afa 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java
@@ -43,7 +43,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.spy;
@@ -178,12 +177,7 @@ public class BufferPoolTest {
public void testCleanupMemoryAvailabilityWaiterOnBlockTimeout() throws Exception {
BufferPool pool = new BufferPool(2, 1, metrics, time, metricGroup);
pool.allocate(1, maxBlockTimeMs);
- try {
- pool.allocate(2, maxBlockTimeMs);
- fail("The buffer allocated more memory than its maximum value 2");
- } catch (BufferExhaustedException e) {
- // this is good
- }
+ assertThrows(BufferExhaustedException.class, () -> pool.allocate(2, maxBlockTimeMs), "The buffer allocated more memory than its maximum value 2");
assertEquals(0, pool.queued());
assertEquals(1, pool.availableMemory());
}
@@ -228,11 +222,7 @@ public class BufferPoolTest {
doThrow(new OutOfMemoryError()).when(bufferPool).recordWaitTime(anyLong());
bufferPool.allocate(1, 0);
- try {
- bufferPool.allocate(2, 1000);
- fail("Expected oom.");
- } catch (OutOfMemoryError expected) {
- }
+ assertThrows(OutOfMemoryError.class, () -> bufferPool.allocate(2, 1000));
assertEquals(1, bufferPool.availableMemory());
assertEquals(0, bufferPool.queued());
assertEquals(1, bufferPool.unallocatedMemory());
@@ -253,14 +243,11 @@ public class BufferPoolTest {
@Override
public void run() {
- try {
- pool.allocate(2, maxBlockTimeMs);
- fail("The buffer allocated more memory than its maximum value 2");
- } catch (BufferExhaustedException e) {
- // this is good
- } catch (InterruptedException e) {
- // this can be neglected
- }
+ Exception e = assertThrows(Exception.class, () -> pool.allocate(2, maxBlockTimeMs));
+ assertTrue(
+ e instanceof BufferExhaustedException || e instanceof InterruptedException,
+ "The buffer allocated more memory than its maximum value 2, exception type: " + e.getClass()
+ );
}
}
@@ -323,14 +310,7 @@ public class BufferPoolTest {
throw new OutOfMemoryError();
}
};
-
- try {
- bufferPool.allocateByteBuffer(1024);
- // should not reach here
- fail("Should have thrown OutOfMemoryError");
- } catch (OutOfMemoryError ignored) {
-
- }
+ assertThrows(OutOfMemoryError.class, () -> bufferPool.allocateByteBuffer(1024));
assertEquals(1024, bufferPool.availableMemory());
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java
index 2e97dcc6c6c..8f78b957546 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java
@@ -100,12 +100,7 @@ public class ProducerBatchTest {
assertEquals(exception, callback.exception);
assertNull(callback.metadata);
- try {
- batch.abort(new KafkaException());
- fail("Expected exception from abort");
- } catch (IllegalStateException e) {
- // expected
- }
+ assertThrows(IllegalStateException.class, () -> batch.abort(new KafkaException()), "Expected exception from abort");
assertEquals(1, callback.invocations);
assertTrue(future.isDone());
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java
index 559ab4d12d5..783534863db 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java
@@ -40,7 +40,6 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class ProducerMetadataTest {
private static final long METADATA_IDLE_MS = 60 * 1000;
@@ -143,20 +142,10 @@ public class ProducerMetadataTest {
metadata.updateWithCurrentRequestVersion(responseWithCurrentTopics(), false, time);
assertTrue(metadata.timeToNextUpdate(time) > 0, "No update needed.");
// first try with a max wait time of 0 and ensure that this returns back without waiting forever
- try {
- metadata.awaitUpdate(metadata.requestUpdate(true), Time.SYSTEM.timer(0));
- fail("Wait on metadata update was expected to timeout, but it didn't");
- } catch (TimeoutException te) {
- // expected
- }
+ assertThrows(TimeoutException.class, () -> metadata.awaitUpdate(metadata.requestUpdate(true), Time.SYSTEM.timer(0)), "Wait on metadata update was expected to timeout, but it didn't");
// now try with a higher timeout value once
final long twoSecondWait = 2000;
- try {
- metadata.awaitUpdate(metadata.requestUpdate(true), Time.SYSTEM.timer(twoSecondWait));
- fail("Wait on metadata update was expected to timeout, but it didn't");
- } catch (TimeoutException te) {
- // expected
- }
+ assertThrows(TimeoutException.class, () -> metadata.awaitUpdate(metadata.requestUpdate(true), Time.SYSTEM.timer(twoSecondWait)), "Wait on metadata update was expected to timeout, but it didn't");
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
index 1708a79b84d..1867aee0a5d 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
@@ -89,7 +89,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class RecordAccumulatorTest {
@@ -699,12 +698,8 @@ public class RecordAccumulatorTest {
accum.beginFlush();
assertTrue(accum.flushInProgress());
delayedInterrupt(Thread.currentThread(), 1000L);
- try {
- accum.awaitFlushCompletion();
- fail("awaitFlushCompletion should throw InterruptException");
- } catch (InterruptedException e) {
- assertFalse(accum.flushInProgress(), "flushInProgress count should be decremented even if thread is interrupted");
- }
+ assertThrows(InterruptedException.class, () -> accum.awaitFlushCompletion());
+ assertFalse(accum.flushInProgress(), "flushInProgress count should be decremented even if thread is interrupted");
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
index 61dcab8ef07..f5944c4db6f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
@@ -121,6 +121,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.apache.kafka.clients.producer.internals.ProducerTestUtils.runUntil;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -131,7 +132,6 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.AdditionalMatchers.geq;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
@@ -3096,12 +3096,8 @@ public class SenderTest {
assertFutureFailure(future2, TransactionAbortableException.class);
// Verify transaction API requests also fail with TransactionAbortableException
- try {
- txnManager.beginCommit();
- fail("Expected beginCommit() to fail with TransactionAbortableException when in abortable error state");
- } catch (KafkaException e) {
- assertEquals(TransactionAbortableException.class, e.getCause().getClass());
- }
+ KafkaException e = assertThrows(KafkaException.class, () -> txnManager.beginCommit(), "Expected beginCommit() to fail with TransactionAbortableException when in abortable error state");
+ assertEquals(TransactionAbortableException.class, e.getCause().getClass());
}
@Test
@@ -3280,13 +3276,9 @@ public class SenderTest {
// Attempt to commit transaction
TransactionalRequestResult commitResult = transactionManager.beginCommit();
sender.runOnce();
- try {
- commitResult.await(1000, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
- fail("Expected abortable error to be thrown for commit");
- } catch (KafkaException e) {
- assertTrue(transactionManager.hasAbortableError());
- assertEquals(TransactionAbortableException.class, commitResult.error().getClass());
- }
+ assertThrows(KafkaException.class, () -> commitResult.await(1000, TimeUnit.MILLISECONDS, "Unexpected time out during the test"), "Expected abortable error to be thrown for commit");
+ assertTrue(transactionManager.hasAbortableError());
+ assertEquals(TransactionAbortableException.class, commitResult.error().getClass());
// Abort API with TRANSACTION_ABORTABLE error should convert to Fatal error i.e. KafkaException
client.prepareResponse(new EndTxnResponse(new EndTxnResponseData()
@@ -3297,15 +3289,11 @@ public class SenderTest {
sender.runOnce();
// Verify the error is converted to KafkaException (not TransactionAbortableException)
- try {
- abortResult.await(1000, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
- fail("Expected KafkaException to be thrown");
- } catch (KafkaException e) {
- // Verify TM is in FATAL_ERROR state
- assertTrue(transactionManager.hasFatalError());
- assertFalse(e instanceof TransactionAbortableException);
- assertEquals(KafkaException.class, abortResult.error().getClass());
- }
+ KafkaException e = assertThrows(KafkaException.class, () -> abortResult.await(1000, TimeUnit.MILLISECONDS, "Unexpected time out during the test"));
+ // Verify TM is in FATAL_ERROR state
+ assertTrue(transactionManager.hasFatalError());
+ assertFalse(e instanceof TransactionAbortableException);
+ assertEquals(KafkaException.class, abortResult.error().getClass());
}
@Test
@@ -3881,23 +3869,15 @@ public class SenderTest {
client.respond(produceResponse(tp0, 0, Errors.NONE, 0));
sender.runOnce();
assertTrue(future.isDone());
- try {
- future.get();
- } catch (ExecutionException e) {
- fail("Future should not have raised an exception: " + e.getCause());
- }
+ assertDoesNotThrow(() -> future.get(), "Future should not have raised an exception");
}
private void assertSendFailure(Class<? extends RuntimeException> expectedError) throws Exception {
Future<RecordMetadata> future = appendToAccumulator(tp0);
sender.runOnce();
assertTrue(future.isDone());
- try {
- future.get();
- fail("Future should have raised " + expectedError.getSimpleName());
- } catch (ExecutionException e) {
- assertTrue(expectedError.isAssignableFrom(e.getCause().getClass()));
- }
+ ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(), "Future should have raised " + expectedError.getSimpleName());
+ assertTrue(expectedError.isAssignableFrom(e.getCause().getClass()));
}
private void prepareAndReceiveInitProducerId(long producerId, Errors error) {
@@ -3948,13 +3928,9 @@ public class SenderTest {
private void assertFutureFailure(Future<?> future, Class<? extends Exception> expectedExceptionType)
throws InterruptedException {
assertTrue(future.isDone());
- try {
- future.get();
- fail("Future should have raised " + expectedExceptionType.getName());
- } catch (ExecutionException e) {
- Class<? extends Throwable> causeType = e.getCause().getClass();
- assertTrue(expectedExceptionType.isAssignableFrom(causeType), "Unexpected cause " + causeType.getName());
- }
+ ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(), "Future should have raised " + expectedExceptionType.getName());
+ Class<? extends Throwable> causeType = e.getCause().getClass();
+ assertTrue(expectedExceptionType.isAssignableFrom(causeType), "Unexpected cause " + causeType.getName());
}
private void createMockClientWithMaxFlightOneMetadataPending() {
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
index 97fe1913fa8..4d754943906 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
@@ -4755,13 +4755,7 @@ public class TransactionManagerTest {
private void assertProduceFutureFailed(Future<RecordMetadata> future) throws InterruptedException {
assertTrue(future.isDone());
-
- try {
- future.get();
- fail("Expected produce future to throw");
- } catch (ExecutionException e) {
- // expected
- }
+ assertThrows(ExecutionException.class, () -> future.get());
}
private void runUntil(Supplier<Boolean> condition) {
diff --git a/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java b/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java
index 1f0a2c794e2..8b4875a6f03 100644
--- a/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java
@@ -47,7 +47,6 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class ConfigDefTest {
@@ -415,12 +414,7 @@ public class ConfigDefTest {
Set<String> names = configDef.names();
assertEquals(Set.of("a", "b"), names);
// should be unmodifiable
- try {
- names.add("new");
- fail();
- } catch (UnsupportedOperationException e) {
- // expected
- }
+ assertThrows(UnsupportedOperationException.class, () -> names.add("new"));
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/common/internals/TopicTest.java b/clients/src/test/java/org/apache/kafka/common/internals/TopicTest.java
index 2eb4a7802a9..0886423ef8d 100644
--- a/clients/src/test/java/org/apache/kafka/common/internals/TopicTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/internals/TopicTest.java
@@ -27,8 +27,8 @@ import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class TopicTest {
@@ -49,12 +49,7 @@ public class TopicTest {
String[] invalidTopicNames = {"", "foo bar", "..", "foo:bar", "foo=bar", ".", new String(longString)};
for (String topicName : invalidTopicNames) {
- try {
- Topic.validate(topicName);
- fail("No exception was thrown for topic with invalid name: " + topicName);
- } catch (InvalidTopicException e) {
- // Good
- }
+ assertThrows(InvalidTopicException.class, () -> Topic.validate(topicName), "No exception was thrown for topic with invalid name: " + topicName);
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java b/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java
index 57bca731177..56db27e011b 100644
--- a/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java
@@ -33,7 +33,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
@Timeout(120)
public class ApiMessageTypeTest {
@@ -48,12 +47,7 @@ public class ApiMessageTypeTest {
@Test
public void testInvalidFromApiKey() {
- try {
- ApiMessageType.fromApiKey((short) -1);
- fail("expected to get an UnsupportedVersionException");
- } catch (UnsupportedVersionException uve) {
- // expected
- }
+ assertThrows(UnsupportedVersionException.class, () -> ApiMessageType.fromApiKey((short) -1));
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/KafkaMbeanTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/KafkaMbeanTest.java
index 6ee50d1623f..b380421f217 100644
--- a/clients/src/test/java/org/apache/kafka/common/metrics/KafkaMbeanTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/metrics/KafkaMbeanTest.java
@@ -74,12 +74,7 @@ public class KafkaMbeanTest {
@Test
public void testGetAttributeUnknown() throws Exception {
sensor.record(2.5);
- try {
- getAttribute(sumMetricName, "name");
- fail("Should have gotten attribute not found");
- } catch (AttributeNotFoundException e) {
- // Expected
- }
+ assertThrows(AttributeNotFoundException.class, () -> getAttribute(sumMetricName, "name"));
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
index 6b806c6bb7b..2e4e8457f89 100644
--- a/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
@@ -45,7 +45,6 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class SensorTest {
@@ -159,12 +158,7 @@ public class SensorTest {
// but adding the same metric to a DIFFERENT sensor is an error
final Sensor anotherSensor = metrics.sensor("another-sensor");
- try {
- anotherSensor.add(metrics.metricName("test-metric", "test-group"), new Avg());
- fail("should have thrown");
- } catch (final IllegalArgumentException ignored) {
- // pass
- }
+ assertThrows(IllegalArgumentException.class, () -> anotherSensor.add(metrics.metricName("test-metric", "test-group"), new Avg()));
// note that adding a different metric with the same name is also a no-op
assertTrue(sensor.add(metrics.metricName("test-metric", "test-group"), new WindowedSum()));
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java b/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
index 12044902582..86d3165c61b 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
@@ -58,8 +58,8 @@ import javax.security.auth.spi.LoginModule;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class SaslChannelBuilderTest {
@@ -92,13 +92,8 @@ public class SaslChannelBuilderTest {
@Test
public void testLoginManagerReleasedIfConfigureThrowsException() {
SaslChannelBuilder builder = createChannelBuilder(SecurityProtocol.SASL_SSL, "PLAIN");
- try {
- // Use invalid config so that an exception is thrown
- builder.configure(Collections.singletonMap(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, "1"));
- fail("Exception should have been thrown");
- } catch (KafkaException e) {
- assertTrue(builder.loginManagers().isEmpty());
- }
+ assertThrows(KafkaException.class, () -> builder.configure(Collections.singletonMap(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, "1")));
+ assertTrue(builder.loginManagers().isEmpty());
builder.close();
assertTrue(builder.loginManagers().isEmpty());
}
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
index 1ea79d1ca1b..e83b1a82a20 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
@@ -158,12 +158,7 @@ public class SelectorTest {
String node = "0";
blockingConnect(node);
selector.send(createSend(node, "test1"));
- try {
- selector.send(createSend(node, "test2"));
- fail("IllegalStateException not thrown when sending a request with one in flight");
- } catch (IllegalStateException e) {
- // Expected exception
- }
+ assertThrows(IllegalStateException.class, () -> selector.send(createSend(node, "test2")), "IllegalStateException not thrown when sending a request with one in flight");
selector.poll(0);
assertTrue(selector.disconnected().containsKey(node), "Channel not closed");
assertEquals(ChannelState.FAILED_SEND, selector.disconnected().get(node));
diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java
index 14bd2ae2982..fd942f218b3 100644
--- a/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java
@@ -31,7 +31,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class ProtocolSerializationTest {
@@ -150,12 +149,19 @@ public class ProtocolSerializationTest {
for (BoundField f : this.schema.fields()) {
Object o = this.struct.get(f);
try {
- this.struct.set(f, null);
- this.struct.validate();
- if (!f.def.type.isNullable())
- fail("Should not allow serialization of null value.");
- } catch (SchemaException e) {
- assertFalse(f.def.type.isNullable(), f + " should not be nullable");
+ if (f.def.type.isNullable()) {
+ this.struct.set(f, null);
+ this.struct.validate();
+ } else {
+ assertThrows(
+ SchemaException.class,
+ () -> {
+ this.struct.set(f, null);
+ this.struct.validate();
+ }, "Should not allow serialization of null value."
+ );
+ assertFalse(f.def.type.isNullable(), f + " should not be nullable");
+ }
} finally {
this.struct.set(f, o);
}
diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/types/RawTaggedFieldWriterTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/types/RawTaggedFieldWriterTest.java
index 78d0c1cf56d..93e394e0665 100644
--- a/clients/src/test/java/org/apache/kafka/common/protocol/types/RawTaggedFieldWriterTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/protocol/types/RawTaggedFieldWriterTest.java
@@ -28,7 +28,7 @@ import java.util.List;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assertions.assertThrows;
@Timeout(120)
public class RawTaggedFieldWriterTest {
@@ -70,12 +70,8 @@ public class RawTaggedFieldWriterTest {
);
RawTaggedFieldWriter writer = RawTaggedFieldWriter.forFields(tags);
assertEquals(3, writer.numFields());
- try {
- writer.writeRawTags(new ByteBufferAccessor(ByteBuffer.allocate(1024)), 2);
- fail("expected to get RuntimeException");
- } catch (RuntimeException e) {
- assertEquals("Attempted to use tag 2 as an undefined tag.", e.getMessage());
- }
+ RuntimeException e = assertThrows(RuntimeException.class, () -> writer.writeRawTags(new ByteBufferAccessor(ByteBuffer.allocate(1024)), 2));
+ assertEquals("Attempted to use tag 2 as an undefined tag.", e.getMessage());
}
@Test
@@ -87,12 +83,7 @@ public class RawTaggedFieldWriterTest {
);
RawTaggedFieldWriter writer = RawTaggedFieldWriter.forFields(tags);
assertEquals(3, writer.numFields());
- try {
- writer.writeRawTags(new ByteBufferAccessor(ByteBuffer.allocate(1024)), 8);
- fail("expected to get RuntimeException");
- } catch (RuntimeException e) {
- assertEquals("Invalid raw tag field list: tag 2 comes after tag 5, but is " +
- "not higher than it.", e.getMessage());
- }
+ RuntimeException e = assertThrows(RuntimeException.class, () -> writer.writeRawTags(new ByteBufferAccessor(ByteBuffer.allocate(1024)), 8));
+ assertEquals("Invalid raw tag field list: tag 2 comes after tag 5, but is " + "not higher than it.", e.getMessage());
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java
index 242896e0fb8..89be472b170 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java
@@ -30,7 +30,6 @@ import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class AbstractLegacyRecordBatchTest {
@@ -223,32 +222,26 @@ public class AbstractLegacyRecordBatchTest {
};
// Check V0
- try {
+ IllegalArgumentException e1 = assertThrows(IllegalArgumentException.class, () -> {
MemoryRecords records = MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V0, 0L,
- Compression.zstd().build(), TimestampType.CREATE_TIME, simpleRecords);
+ Compression.zstd().build(), TimestampType.CREATE_TIME, simpleRecords);
ByteBufferLegacyRecordBatch batch = new ByteBufferLegacyRecordBatch(records.buffer());
batch.setLastOffset(1L);
-
batch.iterator();
- fail("Can't reach here");
- } catch (IllegalArgumentException e) {
- assertEquals("ZStandard compression is not supported for magic 0", e.getMessage());
- }
+ });
+ assertEquals("ZStandard compression is not supported for magic 0", e1.getMessage());
// Check V1
- try {
+ IllegalArgumentException e2 = assertThrows(IllegalArgumentException.class, () -> {
MemoryRecords records = MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V1, 0L,
- Compression.zstd().build(), TimestampType.CREATE_TIME, simpleRecords);
+ Compression.zstd().build(), TimestampType.CREATE_TIME, simpleRecords);
ByteBufferLegacyRecordBatch batch = new ByteBufferLegacyRecordBatch(records.buffer());
batch.setLastOffset(1L);
-
batch.iterator();
- fail("Can't reach here");
- } catch (IllegalArgumentException e) {
- assertEquals("ZStandard compression is not supported for magic 1", e.getMessage());
- }
+ });
+ assertEquals("ZStandard compression is not supported for magic 1", e2.getMessage());
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
index f8845d5c987..cfea14cf708 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
@@ -53,7 +53,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -335,12 +334,7 @@ public class FileRecordsTest {
FileRecords fileRecords = new FileRecords(tempFile(), channelMock, Integer.MAX_VALUE);
- try {
- fileRecords.truncateTo(43);
- fail("Should throw KafkaException");
- } catch (KafkaException e) {
- // expected
- }
+ assertThrows(KafkaException.class, () -> fileRecords.truncateTo(43));
verify(channelMock, atLeastOnce()).size();
}
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/LeaveGroupRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/LeaveGroupRequestTest.java
index 82e647128f9..1602e336081 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/LeaveGroupRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/LeaveGroupRequestTest.java
@@ -33,7 +33,6 @@ import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class LeaveGroupRequestTest {
@@ -66,26 +65,28 @@ public class LeaveGroupRequestTest {
.setMembers(members);
for (short version : ApiKeys.LEAVE_GROUP.allVersions()) {
- try {
+ if (version <= 2) {
+ UnsupportedVersionException e = assertThrows(
+ UnsupportedVersionException.class,
+ () -> builder.build(version), "Older version " + version + " request data should not be created due to non-single members"
+ );
+ assertTrue(e.getMessage().contains("leave group request only supports single member instance"));
+ } else {
LeaveGroupRequest request = builder.build(version);
- if (version <= 2) {
- fail("Older version " + version +
- " request data should not be created due to non-single members");
- }
assertEquals(expectedData, request.data());
assertEquals(members, request.members());
LeaveGroupResponse expectedResponse = new LeaveGroupResponse(
- Collections.emptyList(),
- Errors.COORDINATOR_LOAD_IN_PROGRESS,
- throttleTimeMs,
- version
+ Collections.emptyList(),
+ Errors.COORDINATOR_LOAD_IN_PROGRESS,
+ throttleTimeMs,
+ version
);
- assertEquals(expectedResponse, request.getErrorResponse(throttleTimeMs,
- Errors.COORDINATOR_LOAD_IN_PROGRESS.exception()));
- } catch (UnsupportedVersionException e) {
- assertTrue(e.getMessage().contains("leave group request only supports single member instance"));
+ assertEquals(
+ expectedResponse,
+ request.getErrorResponse(throttleTimeMs, Errors.COORDINATOR_LOAD_IN_PROGRESS.exception())
+ );
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java b/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java
index 760b1afc41f..eb060afca04 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java
@@ -45,7 +45,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
/**
* Tests parsing of {@link SaslConfigs#SASL_JAAS_CONFIG} property and verifies that the format
@@ -394,18 +393,11 @@ public class JaasContextTest {
}
private void checkInvalidConfiguration(String jaasConfigProp) throws IOException {
- try {
+ assertThrows(SecurityException.class, () -> {
writeConfiguration(JaasContext.Type.SERVER.name(), jaasConfigProp);
- AppConfigurationEntry entry = configurationEntry(JaasContext.Type.SERVER, null);
- fail("Invalid JAAS configuration file didn't throw exception, entry=" + entry);
- } catch (SecurityException e) {
- // Expected exception
- }
- try {
- AppConfigurationEntry entry = configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp);
- fail("Invalid JAAS configuration property didn't throw exception, entry=" + entry);
- } catch (IllegalArgumentException e) {
- // Expected exception
- }
+ configurationEntry(JaasContext.Type.SERVER, null);
+ }, "Invalid JAAS configuration file didn't throw exception");
+ assertThrows(IllegalArgumentException.class, () -> configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp), "Invalid JAAS configuration property didn't throw exception");
+
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
index fbb4f378c89..5175f76c3c3 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
@@ -360,12 +360,7 @@ public class SaslAuthenticatorTest {
server = createEchoServer(securityProtocol);
createSelector(securityProtocol, saslClientConfigs);
InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
- try {
- selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
- fail("SASL/PLAIN channel created without password");
- } catch (IOException e) {
- // Expected exception
- }
+ assertThrows(IOException.class, () -> selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE), "SASL/PLAIN channel created without password");
}
/**
@@ -1084,12 +1079,7 @@ public class SaslAuthenticatorTest {
SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
server = createEchoServer(securityProtocol);
- try {
- createSelector(securityProtocol, saslClientConfigs);
- fail("SASL/PLAIN channel created without valid login module");
- } catch (KafkaException e) {
- // Expected exception
- }
+ assertThrows(KafkaException.class, () -> createSelector(securityProtocol, saslClientConfigs), "SASL/PLAIN channel created without valid login module");
}
/**
@@ -1253,13 +1243,7 @@ public class SaslAuthenticatorTest {
saslServerConfigs.put(prefix + BrokerSecurityConfigs.SASL_SERVER_CALLBACK_HANDLER_CLASS_CONFIG,
TestServerCallbackHandler.class);
Class<?> loginCallback = TestLoginCallbackHandler.class;
-
- try {
- createEchoServer(securityProtocol);
- fail("Should have failed to create server with default login handler");
- } catch (KafkaException e) {
- // Expected exception
- }
+ assertThrows(KafkaException.class, () -> createEchoServer(securityProtocol), "Should have failed to create server with default login handler");
try {
saslServerConfigs.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, loginCallback);
@@ -1374,12 +1358,7 @@ public class SaslAuthenticatorTest {
String module1 = TestJaasConfig.jaasConfigProperty("PLAIN", "user1", "user1-secret").value();
String module2 = TestJaasConfig.jaasConfigProperty("PLAIN", "user2", "user2-secret").value();
saslClientConfigs.put(SaslConfigs.SASL_JAAS_CONFIG, new Password(module1 + " " + module2));
- try {
- createClientConnection(securityProtocol, "1");
- fail("Connection created with multiple login modules in sasl.jaas.config");
- } catch (IllegalArgumentException e) {
- // Expected
- }
+ assertThrows(IllegalArgumentException.class, () -> createClientConnection(securityProtocol, "1"), "Connection created with multiple login modules in sasl.jaas.config");
}
/**
diff --git a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerScopeUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerScopeUtilsTest.java
index 7cf75406134..d78a083ebb3 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerScopeUtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerScopeUtilsTest.java
@@ -22,8 +22,8 @@ import org.junit.jupiter.api.Test;
import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class OAuthBearerScopeUtilsTest {
@Test
@@ -44,12 +44,7 @@ public class OAuthBearerScopeUtilsTest {
@Test
public void invalidScope() {
for (String invalidScope : new String[] {"\"foo", "\\foo"}) {
- try {
- OAuthBearerScopeUtils.parseScope(invalidScope);
- fail("did not detect invalid scope: " + invalidScope);
- } catch (OAuthBearerConfigException expected) {
- // empty
- }
+ assertThrows(OAuthBearerConfigException.class, () -> OAuthBearerScopeUtils.parseScope(invalidScope), "did not detect invalid scope: " + invalidScope);
}
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/security/scram/internals/ScramMessagesTest.java b/clients/src/test/java/org/apache/kafka/common/security/scram/internals/ScramMessagesTest.java
index df84c054d45..a2cd28465a3 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/scram/internals/ScramMessagesTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/scram/internals/ScramMessagesTest.java
@@ -34,8 +34,8 @@ import javax.security.sasl.SaslException;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public class ScramMessagesTest {
@@ -344,11 +344,6 @@ public class ScramMessagesTest {
}
private <T extends AbstractScramMessage> void checkInvalidScramMessage(Class<T> clazz, String message) {
- try {
- createScramMessage(clazz, message);
- fail("Exception not throws for invalid message of type " + clazz + " : " + message);
- } catch (SaslException e) {
- // Expected exception
- }
+ assertThrows(SaslException.class, () -> createScramMessage(clazz, message), "Exception not throws for invalid message of type " + clazz + " : " + message);
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
index 9e888c6ed6c..ccc250d2d55 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
@@ -275,12 +275,7 @@ public abstract class SslFactoryTest {
Map<String, Object> sslConfig2 = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
- try {
- sslFactory.validateReconfiguration(sslConfig2);
- fail("Truststore configured dynamically for listener without previous truststore");
- } catch (ConfigException e) {
- // Expected exception
- }
+ assertThrows(ConfigException.class, () -> sslFactory.validateReconfiguration(sslConfig2), "Truststore configured dynamically for listener without previous truststore");
}
@Test
@@ -311,15 +306,10 @@ public abstract class SslFactoryTest {
assertNotSame(sslContext, ((DefaultSslEngineFactory) sslFactory.sslEngineFactory()).sslContext(),
"SSL context not recreated");
- sslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
+ Map<String, Object> sslConfig2 = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(newTrustStoreFile)
.build();
- try {
- sslFactory.validateReconfiguration(sslConfig);
- fail("Keystore configured dynamically for listener without previous keystore");
- } catch (ConfigException e) {
- // Expected exception
- }
+ assertThrows(ConfigException.class, () -> sslFactory.validateReconfiguration(sslConfig2), "Keystore configured dynamically for listener without previous keystore");
}
@Test
@@ -415,12 +405,7 @@ public abstract class SslFactoryTest {
SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG)) {
sslConfig1.put(key, sslConfig2.get(key));
}
- try {
- sslFactory.configure(sslConfig1);
- fail("Validation did not fail with untrusted truststore");
- } catch (ConfigException e) {
- // Expected exception
- }
+ assertThrows(ConfigException.class, () -> sslFactory.configure(sslConfig1), "Validation did not fail with untrusted truststore");
}
@Test
@@ -451,12 +436,7 @@ public abstract class SslFactoryTest {
// the new truststore, if certificate is not trusted by the existing truststore on the `SslFactory`.
// This is to prevent both keystores and truststores to be modified simultaneously on an inter-broker
// listener to stores that may not work with other brokers where the update hasn't yet been performed.
- try {
- sslFactory.validateReconfiguration(sslConfig2);
- fail("ValidateReconfiguration did not fail as expected");
- } catch (ConfigException e) {
- // Expected exception
- }
+ assertThrows(ConfigException.class, () -> sslFactory.validateReconfiguration(sslConfig2), "ValidateReconfiguration did not fail as expected");
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
index 413fd6f7d81..70d9c220058 100755
--- a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
@@ -604,12 +604,7 @@ public class UtilsTest {
assertFalse(smallBuffer.hasRemaining(), "Buffer should be filled");
assertEquals("world", new String(smallBuffer.array()), "Buffer should be populated correctly");
// Scenario 4: test end of stream is reached before buffer is filled up
- try {
- Utils.readFullyOrFail(channel, largeBuffer, 0, "large");
- fail("Expected EOFException to be raised");
- } catch (EOFException e) {
- // expected
- }
+ assertThrows(EOFException.class, () -> Utils.readFullyOrFail(channel, largeBuffer, 0, "large"));
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/internals/ImplicitLinkedHashCollectionTest.java b/clients/src/test/java/org/apache/kafka/common/utils/internals/ImplicitLinkedHashCollectionTest.java
index fc43fa94878..49ac3279ce5 100644
--- a/clients/src/test/java/org/apache/kafka/common/utils/internals/ImplicitLinkedHashCollectionTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/internals/ImplicitLinkedHashCollectionTest.java
@@ -36,7 +36,6 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
/**
* A unit test for ImplicitLinkedHashCollection.
@@ -398,12 +397,7 @@ public class ImplicitLinkedHashCollectionTest {
coll.add(new TestElement(5));
ListIterator<TestElement> iter = coll.valuesList().listIterator();
- try {
- iter.remove();
- fail("Calling remove() without calling next() or previous() should raise an exception");
- } catch (IllegalStateException e) {
- // expected
- }
+ assertThrows(IllegalStateException.class, () -> iter.remove(), "Calling remove() without calling next() or previous() should raise an exception");
// Remove after next()
iter.next();
@@ -415,12 +409,7 @@ public class ImplicitLinkedHashCollectionTest {
assertEquals(2, iter.nextIndex());
assertEquals(1, iter.previousIndex());
- try {
- iter.remove();
- fail("Calling remove() twice without calling next() or previous() in between should raise an exception");
- } catch (IllegalStateException e) {
- // expected
- }
+ assertThrows(IllegalStateException.class, () -> iter.remove(), "Calling remove() twice without calling next() or previous() in between should raise an exception");
// Remove after previous()
assertEquals(2, iter.previous().key);