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/16 06:48:21 UTC
(kafka) branch trunk updated: MINOR: Replace try/fail/catch with assertThrows in streams and kstream tests (#23389)
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 56baca2b652 MINOR: Replace try/fail/catch with assertThrows in streams and kstream tests (#23389)
56baca2b652 is described below
commit 56baca2b65251318a17154034a638b6f37654c26
Author: JunHwan <de...@gmail.com>
AuthorDate: Wed Sep 16 15:48:14 2026 +0900
MINOR: Replace try/fail/catch with assertThrows in streams and kstream tests (#23389)
Replace the remaining straightforward try/fail/catch exception
assertions in the `org.apache.kafka.streams` and
`org.apache.kafka.streams.kstream` test packages with `assertThrows`,
same as #23277.
Where the catch block checked the exception message, the `assertThrows`
return value is used to keep that check. Unused `fail` imports are
removed.
Preserve descriptive failure messages from the original `fail()` calls
and clarify generic messages to identify rejected operations and invalid
inputs.
Files in `streams.internals`, `streams.state.internals`,
`streams.integration`, and `streams.processor.internals` are left
unchanged because they overlap with the in-flight Hamcrest removal work
(#23307, #23386, #23387, and KAFKA-21028, which is in progress).
`KafkaStreamsTest#shouldThrowTopologyExceptionOnEmptyTopology` is also
left as is, since its `fail()` sits inside a try-with-resources block
that closes the `KafkaStreams` instance.
Testing:
Validated on Windows with Java 25 and Gradle 9.7.1. All 426 tests in the
seven affected test classes passed, along with Streams Checkstyle and
Spotless checks.
Equivalent Gradle wrapper commands:
```sh
./gradlew :streams:checkstyleMain :streams:checkstyleTest
:streams:spotlessCheck
./gradlew :streams:test \
--tests org.apache.kafka.streams.TopologyTest \
--tests org.apache.kafka.streams.StreamsConfigTest \
--tests org.apache.kafka.streams.KafkaStreamsTest \
--tests org.apache.kafka.streams.kstream.TimeWindowsTest \
--tests org.apache.kafka.streams.kstream.JoinWindowsTest \
--tests org.apache.kafka.streams.kstream.SessionWindowsTest \
--tests org.apache.kafka.streams.kstream.NamedTest
```
Reviewers: Ken Huang <s7...@gmail.com>
---------
Co-authored-by: Claude Fable 5.1 <no...@anthropic.com>
---
.../org/apache/kafka/streams/KafkaStreamsTest.java | 45 ++++----
.../apache/kafka/streams/StreamsConfigTest.java | 94 +++++++--------
.../org/apache/kafka/streams/TopologyTest.java | 127 ++++++++++++---------
.../kafka/streams/kstream/JoinWindowsTest.java | 34 +++---
.../apache/kafka/streams/kstream/NamedTest.java | 14 +--
.../kafka/streams/kstream/SessionWindowsTest.java | 12 +-
.../kafka/streams/kstream/TimeWindowsTest.java | 45 ++++----
7 files changed, 184 insertions(+), 187 deletions(-)
diff --git a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
index b2c170c3de5..c60c2746feb 100644
--- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
@@ -1022,12 +1022,11 @@ public class KafkaStreamsTest {
try (final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time)) {
streams.start();
streams.close();
- try {
- streams.start();
- fail("Should have throw IllegalStateException");
- } catch (final IllegalStateException expected) {
- // this is ok
- }
+ assertThrows(
+ IllegalStateException.class,
+ streams::start,
+ "should not allow restarting after close"
+ );
}
}
@@ -1040,12 +1039,11 @@ public class KafkaStreamsTest {
prepareThreadState(streamThreadTwo, state2);
try (final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time)) {
streams.start();
- try {
- streams.setGlobalStateRestoreListener(null);
- fail("Should throw an IllegalStateException");
- } catch (final IllegalStateException e) {
- // expected
- }
+ assertThrows(
+ IllegalStateException.class,
+ () -> streams.setGlobalStateRestoreListener(null),
+ "should not allow setting the global state restore listener after start"
+ );
}
}
@@ -1090,12 +1088,11 @@ public class KafkaStreamsTest {
prepareStreamThread(streamThreadTwo, 2);
try (final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time)) {
streams.start();
- try {
- streams.setStateListener(null);
- fail("Should throw IllegalStateException");
- } catch (final IllegalStateException e) {
- // expected
- }
+ assertThrows(
+ IllegalStateException.class,
+ () -> streams.setStateListener(null),
+ "should not allow setting the state listener after start"
+ );
}
}
@@ -1128,12 +1125,12 @@ public class KafkaStreamsTest {
() -> streams.state() == KafkaStreams.State.RUNNING,
"Streams never started.");
- try {
- streams.cleanUp();
- fail("Should have thrown IllegalStateException");
- } catch (final IllegalStateException expected) {
- assertEquals("Cannot clean up while running.", expected.getMessage());
- }
+ final IllegalStateException expected = assertThrows(
+ IllegalStateException.class,
+ streams::cleanUp,
+ "should not allow cleanup while running"
+ );
+ assertEquals("Cannot clean up while running.", expected.getMessage());
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index ac45d48961c..aef7604b68d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -882,15 +882,15 @@ public class StreamsConfigTest {
public void shouldThrowExceptionIfCommitIntervalMsIsNegative() {
final long commitIntervalMs = -1;
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, commitIntervalMs);
- try {
- new StreamsConfig(props);
- fail("Should throw ConfigException when commitIntervalMs is set to a negative value");
- } catch (final ConfigException e) {
- assertEquals(
- "Invalid value -1 for configuration commit.interval.ms: Value must be at least 0",
- e.getMessage()
- );
- }
+ final ConfigException e = assertThrows(
+ ConfigException.class,
+ () -> new StreamsConfig(props),
+ "should not accept a negative commit interval"
+ );
+ assertEquals(
+ "Invalid value -1 for configuration commit.interval.ms: Value must be at least 0",
+ e.getMessage()
+ );
}
@Test
@@ -921,15 +921,15 @@ public class StreamsConfigTest {
final Properties props = getStreamsConfig();
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, MisconfiguredSerde.class);
final StreamsConfig config = new StreamsConfig(props);
- try {
- config.defaultKeySerde();
- fail("Test should throw a StreamsException");
- } catch (final StreamsException e) {
- assertEquals(
- "Failed to configure key serde class org.apache.kafka.streams.StreamsConfigTest$MisconfiguredSerde",
- e.getMessage()
- );
- }
+ final StreamsException e = assertThrows(
+ StreamsException.class,
+ config::defaultKeySerde,
+ "should not return a default key serde that fails to configure"
+ );
+ assertEquals(
+ "Failed to configure key serde class org.apache.kafka.streams.StreamsConfigTest$MisconfiguredSerde",
+ e.getMessage()
+ );
}
@SuppressWarnings("resource")
@@ -938,15 +938,15 @@ public class StreamsConfigTest {
final Properties props = getStreamsConfig();
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, MisconfiguredSerde.class);
final StreamsConfig config = new StreamsConfig(props);
- try {
- config.defaultValueSerde();
- fail("Test should throw a StreamsException");
- } catch (final StreamsException e) {
- assertEquals(
- "Failed to configure value serde class org.apache.kafka.streams.StreamsConfigTest$MisconfiguredSerde",
- e.getMessage()
- );
- }
+ final StreamsException e = assertThrows(
+ StreamsException.class,
+ config::defaultValueSerde,
+ "should not return a default value serde that fails to configure"
+ );
+ assertEquals(
+ "Failed to configure value serde class org.apache.kafka.streams.StreamsConfigTest$MisconfiguredSerde",
+ e.getMessage()
+ );
}
@Test
@@ -954,16 +954,16 @@ public class StreamsConfigTest {
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE_V2);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 7);
final StreamsConfig streamsConfig = new StreamsConfig(props);
- try {
- streamsConfig.getProducerConfigs(clientId);
- fail("Should throw ConfigException when ESO is enabled and maxInFlight requests exceeds 5");
- } catch (final ConfigException e) {
- assertEquals(
- "Invalid value 7 for configuration max.in.flight.requests.per.connection:" +
- " Can't exceed 5 when exactly-once processing is enabled",
- e.getMessage()
- );
- }
+ final ConfigException e = assertThrows(
+ ConfigException.class,
+ () -> streamsConfig.getProducerConfigs(clientId),
+ "should not accept max in flight requests above 5 when exactly-once is enabled"
+ );
+ assertEquals(
+ "Invalid value 7 for configuration max.in.flight.requests.per.connection:" +
+ " Can't exceed 5 when exactly-once processing is enabled",
+ e.getMessage()
+ );
}
@Test
@@ -979,16 +979,16 @@ public class StreamsConfigTest {
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE_V2);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "not-a-number");
- try {
- new StreamsConfig(props).getProducerConfigs(clientId);
- fail("Should throw ConfigException when EOS is enabled and maxInFlight cannot be parsed into an integer");
- } catch (final ConfigException e) {
- assertEquals(
- "Invalid value not-a-number for configuration max.in.flight.requests.per.connection:" +
- " String value could not be parsed as 32-bit integer",
- e.getMessage()
- );
- }
+ final ConfigException e = assertThrows(
+ ConfigException.class,
+ () -> new StreamsConfig(props).getProducerConfigs(clientId),
+ "should not accept a non-numeric max in flight requests when exactly-once is enabled"
+ );
+ assertEquals(
+ "Invalid value not-a-number for configuration max.in.flight.requests.per.connection:" +
+ " String value could not be parsed as 32-bit integer",
+ e.getMessage()
+ );
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
index e4d102f7a40..4eddf9eb0e8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
@@ -85,7 +85,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;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -188,65 +187,72 @@ public class TopologyTest {
@Test
public void shouldNotAllowToAddSourcesWithSameName() {
topology.addSource("source", "topic-1");
- try {
- topology.addSource("source", "topic-2");
- fail("Should throw TopologyException for duplicate source name");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addSource("source", "topic-2"),
+ "should not allow a duplicate source name"
+ );
}
@Test
public void shouldNotAllowToAddTopicTwice() {
topology.addSource("source", "topic-1");
- try {
- topology.addSource("source-2", "topic-1");
- fail("Should throw TopologyException for already used topic");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addSource("source-2", "topic-1"),
+ "should not allow a topic that is already used"
+ );
}
@Test
public void testPatternMatchesAlreadyProvidedTopicSource() {
topology.addSource("source-1", "foo");
- try {
- topology.addSource("source-2", Pattern.compile("f.*"));
- fail("Should have thrown TopologyException for overlapping pattern with already registered topic");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addSource("source-2", Pattern.compile("f.*")),
+ "should not allow a pattern overlapping an already registered topic"
+ );
}
@Test
public void testNamedTopicMatchesAlreadyProvidedPattern() {
topology.addSource("source-1", Pattern.compile("f.*"));
- try {
- topology.addSource("source-2", "foo");
- fail("Should have thrown TopologyException for overlapping topic with already registered pattern");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addSource("source-2", "foo"),
+ "should not allow a topic overlapping an already registered pattern"
+ );
}
@Test
public void shouldNotAllowToAddProcessorWithSameName() {
topology.addSource("source", "topic-1");
topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
- try {
- topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
- fail("Should throw TopologyException for duplicate processor name");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source"),
+ "should not allow a duplicate processor name"
+ );
}
@Test
public void shouldNotAllowToAddProcessorWithEmptyParents() {
topology.addSource("source", "topic-1");
- try {
- topology.addProcessor("processor", new MockApiProcessorSupplier<>());
- fail("Should throw TopologyException for processor without at least one parent node");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addProcessor("processor", new MockApiProcessorSupplier<>()),
+ "should not allow a processor without at least one parent node"
+ );
}
@Test
public void shouldNotAllowToAddProcessorWithNullParents() {
topology.addSource("source", "topic-1");
- try {
- topology.addProcessor("processor", new MockApiProcessorSupplier<>(), (String) null);
- fail("Should throw NullPointerException for processor when null parent names are provided");
- } catch (final NullPointerException expected) { }
+ assertThrows(
+ NullPointerException.class,
+ () -> topology.addProcessor("processor", new MockApiProcessorSupplier<>(), (String) null),
+ "should not allow null parent names for a processor"
+ );
}
@Test
@@ -263,30 +269,33 @@ public class TopologyTest {
public void shouldNotAllowToAddSinkWithSameName() {
topology.addSource("source", "topic-1");
topology.addSink("sink", "topic-2", "source");
- try {
- topology.addSink("sink", "topic-3", "source");
- fail("Should throw TopologyException for duplicate sink name");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addSink("sink", "topic-3", "source"),
+ "should not allow a duplicate sink name"
+ );
}
@Test
public void shouldNotAllowToAddSinkWithEmptyParents() {
topology.addSource("source", "topic-1");
topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
- try {
- topology.addSink("sink", "topic-2");
- fail("Should throw TopologyException for sink without at least one parent node");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addSink("sink", "topic-2"),
+ "should not allow a sink without at least one parent node"
+ );
}
@Test
public void shouldNotAllowToAddSinkWithNullParents() {
topology.addSource("source", "topic-1");
topology.addProcessor("processor", new MockApiProcessorSupplier<>(), "source");
- try {
- topology.addSink("sink", "topic-2", (String) null);
- fail("Should throw NullPointerException for sink when null parent names are provided");
- } catch (final NullPointerException expected) { }
+ assertThrows(
+ NullPointerException.class,
+ () -> topology.addSink("sink", "topic-2", (String) null),
+ "should not allow null parent names for a sink"
+ );
}
@Test
@@ -303,10 +312,11 @@ public class TopologyTest {
public void shouldFailIfSinkIsParent() {
topology.addSource("source", "topic-1");
topology.addSink("sink-1", "topic-2", "source");
- try {
- topology.addSink("sink-2", "topic-3", "sink-1");
- fail("Should throw TopologyException for using sink as parent");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addSink("sink-2", "topic-3", "sink-1"),
+ "should not allow a sink as a parent node"
+ );
}
@Test
@@ -319,10 +329,11 @@ public class TopologyTest {
public void shouldNotAllowToAddStateStoreToSource() {
mockStoreBuilder();
topology.addSource("source-1", "topic-1");
- try {
- topology.addStateStore(storeBuilder, "source-1");
- fail("Should have thrown TopologyException for adding store to source node");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addStateStore(storeBuilder, "source-1"),
+ "should not allow adding a state store to a source node"
+ );
}
@Test
@@ -330,10 +341,11 @@ public class TopologyTest {
mockStoreBuilder();
topology.addSource("source-1", "topic-1");
topology.addSink("sink-1", "topic-1", "source-1");
- try {
- topology.addStateStore(storeBuilder, "sink-1");
- fail("Should have thrown TopologyException for adding store to sink node");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addStateStore(storeBuilder, "sink-1"),
+ "should not allow adding a state store to a sink node"
+ );
}
private void mockStoreBuilder() {
@@ -347,10 +359,11 @@ public class TopologyTest {
final StoreBuilder<?> otherStoreBuilder = mock(StoreBuilder.class);
when(otherStoreBuilder.name()).thenReturn("store");
- try {
- topology.addStateStore(otherStoreBuilder);
- fail("Should have thrown TopologyException for same store name with different StoreBuilder");
- } catch (final TopologyException expected) { }
+ assertThrows(
+ TopologyException.class,
+ () -> topology.addStateStore(otherStoreBuilder),
+ "should not allow the same store name with a different StoreBuilder"
+ );
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java
index ef5188b0f21..db9cad76f76 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java
@@ -27,7 +27,6 @@ import static org.apache.kafka.streams.EqualityCheck.verifyInEquality;
import static org.apache.kafka.streams.kstream.Windows.DEPRECATED_DEFAULT_24_HR_GRACE_PERIOD;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.fail;
public class JoinWindowsTest {
@@ -82,23 +81,21 @@ public class JoinWindowsTest {
@Test
public void endTimeShouldNotBeBeforeStart() {
final JoinWindows windowSpec = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(ANY_SIZE));
- try {
- windowSpec.after(ofMillis(-ANY_SIZE - 1));
- fail("window end time should not be before window start time");
- } catch (final IllegalArgumentException e) {
- // expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> windowSpec.after(ofMillis(-ANY_SIZE - 1)),
+ "should not allow window end time before window start time"
+ );
}
@Test
public void startTimeShouldNotBeAfterEnd() {
final JoinWindows windowSpec = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(ANY_SIZE));
- try {
- windowSpec.before(ofMillis(-ANY_SIZE - 1));
- fail("window start time should not be after window end time");
- } catch (final IllegalArgumentException e) {
- // expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> windowSpec.before(ofMillis(-ANY_SIZE - 1)),
+ "should not allow window start time after window end time"
+ );
}
@SuppressWarnings("deprecation")
@@ -113,12 +110,11 @@ public class JoinWindowsTest {
public void gracePeriodShouldEnforceBoundaries() {
JoinWindows.ofTimeDifferenceAndGrace(ofMillis(3L), ofMillis(0L));
- try {
- JoinWindows.ofTimeDifferenceAndGrace(ofMillis(3L), ofMillis(-1L));
- fail("should not accept negatives");
- } catch (final IllegalArgumentException e) {
- //expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> JoinWindows.ofTimeDifferenceAndGrace(ofMillis(3L), ofMillis(-1L)),
+ "should not accept a negative grace period"
+ );
}
@SuppressWarnings("deprecation")
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/NamedTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/NamedTest.java
index 0a2d735dea9..36915528941 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/NamedTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/NamedTest.java
@@ -23,7 +23,6 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.fail;
public class NamedTest {
@@ -39,12 +38,11 @@ public class NamedTest {
final String[] invalidNames = {"", "foo bar", "..", "foo:bar", "foo=bar", ".", new String(longString)};
for (final String name : invalidNames) {
- try {
- Named.validate(name);
- fail("No exception was thrown for named with invalid name: " + name);
- } catch (final TopologyException e) {
- // success
- }
+ assertThrows(
+ TopologyException.class,
+ () -> Named.validate(name),
+ "should not accept invalid name: '" + name + "'"
+ );
}
}
-}
\ No newline at end of file
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java
index 9b2fd6bb547..5a5a0cdc3d0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java
@@ -23,7 +23,6 @@ import static org.apache.kafka.streams.EqualityCheck.verifyEquality;
import static org.apache.kafka.streams.EqualityCheck.verifyInEquality;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.fail;
public class SessionWindowsTest {
@@ -43,12 +42,11 @@ public class SessionWindowsTest {
public void gracePeriodShouldEnforceBoundaries() {
SessionWindows.ofInactivityGapAndGrace(ofMillis(3L), ofMillis(0));
- try {
- SessionWindows.ofInactivityGapAndGrace(ofMillis(3L), ofMillis(-1L));
- fail("should not accept negatives");
- } catch (final IllegalArgumentException e) {
- //expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> SessionWindows.ofInactivityGapAndGrace(ofMillis(3L), ofMillis(-1L)),
+ "should not accept a negative grace period"
+ );
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java
index 5f3b2a5d39f..8126d86e9cd 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java
@@ -28,7 +28,6 @@ import static org.apache.kafka.streams.EqualityCheck.verifyInEquality;
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.fail;
public class TimeWindowsTest {
@@ -60,46 +59,42 @@ public class TimeWindowsTest {
@Test
public void advanceIntervalMustNotBeZero() {
final TimeWindows windowSpec = TimeWindows.ofSizeWithNoGrace(ofMillis(ANY_SIZE));
- try {
- windowSpec.advanceBy(ofMillis(0));
- fail("should not accept zero advance parameter");
- } catch (final IllegalArgumentException e) {
- // expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> windowSpec.advanceBy(ofMillis(0)),
+ "should not accept zero advance parameter"
+ );
}
@Test
public void advanceIntervalMustNotBeNegative() {
final TimeWindows windowSpec = TimeWindows.ofSizeWithNoGrace(ofMillis(ANY_SIZE));
- try {
- windowSpec.advanceBy(ofMillis(-1));
- fail("should not accept negative advance parameter");
- } catch (final IllegalArgumentException e) {
- // expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> windowSpec.advanceBy(ofMillis(-1)),
+ "should not accept negative advance parameter"
+ );
}
@Test
public void advanceIntervalMustNotBeLargerThanWindowSize() {
final TimeWindows windowSpec = TimeWindows.ofSizeWithNoGrace(ofMillis(ANY_SIZE));
- try {
- windowSpec.advanceBy(ofMillis(ANY_SIZE + 1));
- fail("should not accept advance greater than window size");
- } catch (final IllegalArgumentException e) {
- // expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> windowSpec.advanceBy(ofMillis(ANY_SIZE + 1)),
+ "should not accept advance greater than window size"
+ );
}
@Test
public void gracePeriodShouldEnforceBoundaries() {
TimeWindows.ofSizeAndGrace(ofMillis(3L), ofMillis(0L));
- try {
- TimeWindows.ofSizeAndGrace(ofMillis(3L), ofMillis(-1L));
- fail("should not accept negatives");
- } catch (final IllegalArgumentException e) {
- //expected
- }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> TimeWindows.ofSizeAndGrace(ofMillis(3L), ofMillis(-1L)),
+ "should not accept a negative grace period"
+ );
}
@Test