Skip to content

Commit 44e3f52

Browse files
garyrussellgaryrussell
committed
Change Some Tests to Parameterized
1 parent ab871ab commit 44e3f52

File tree

2 files changed

+33
-61
lines changed

2 files changed

+33
-61
lines changed

‎spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

Lines changed: 32 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import java.util.function.Supplier;
5959
import java.util.regex.Pattern;
6060
import java.util.stream.Collectors;
61+
import java.util.stream.Stream;
6162

6263
import org.aopalliance.intercept.MethodInterceptor;
6364
import org.apache.commons.logging.LogFactory;
@@ -83,6 +84,10 @@
8384
import org.assertj.core.api.InstanceOfAssertFactories;
8485
import org.junit.jupiter.api.BeforeAll;
8586
import org.junit.jupiter.api.Test;
87+
import org.junit.jupiter.params.ParameterizedTest;
88+
import org.junit.jupiter.params.provider.Arguments;
89+
import org.junit.jupiter.params.provider.EnumSource;
90+
import org.junit.jupiter.params.provider.MethodSource;
8691
import org.mockito.ArgumentCaptor;
8792
import org.mockito.InOrder;
8893

@@ -657,18 +662,10 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
657662
assertThat(container.isRunning()).isFalse();
658663
}
659664

660-
@Test
661-
void testInOrderAckManual() throws Exception {
662-
testInOrderAck(AckMode.MANUAL);
663-
}
664-
665-
@Test
666-
void testInOrderAckManualImm() throws Exception {
667-
testInOrderAck(AckMode.MANUAL_IMMEDIATE);
668-
}
669-
665+
@ParameterizedTest(name = "{index} AckMode.{0}")
666+
@EnumSource(value = AckMode.class, names = { "MANUAL", "MANUAL_IMMEDIATE" })
670667
@SuppressWarnings("unchecked")
671-
private void testInOrderAck(AckMode ackMode) throws Exception {
668+
void testInOrderAck(AckMode ackMode) throws Exception {
672669
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
673670
Consumer<Integer, String> consumer = mock(Consumer.class);
674671
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
@@ -723,28 +720,18 @@ private void testInOrderAck(AckMode ackMode) throws Exception {
723720
container.stop();
724721
}
725722

726-
@Test
727-
void testInOrderAckPauseUntilAckedManual() throws Exception {
728-
testInOrderAckPauseUntilAcked(AckMode.MANUAL, false);
729-
}
730-
731-
@Test
732-
void testInOrderAckPauseUntilAckedManualImm() throws Exception {
733-
testInOrderAckPauseUntilAcked(AckMode.MANUAL_IMMEDIATE, false);
734-
}
735-
736-
@Test
737-
void testInOrderAckPauseUntilAckedManualBatch() throws Exception {
738-
testInOrderAckPauseUntilAcked(AckMode.MANUAL, true);
739-
}
740-
741-
@Test
742-
void testInOrderAckPauseUntilAckedManualImmBatch() throws Exception {
743-
testInOrderAckPauseUntilAcked(AckMode.MANUAL_IMMEDIATE, true);
723+
private static Stream<Arguments> testInOrderAckPauseUntilAckedParamters() {
724+
return Stream.of(
725+
Arguments.of(AckMode.MANUAL, false),
726+
Arguments.of(AckMode.MANUAL, true),
727+
Arguments.of(AckMode.MANUAL_IMMEDIATE, false),
728+
Arguments.of(AckMode.MANUAL_IMMEDIATE, true));
744729
}
745730

731+
@ParameterizedTest(name = "{index} AckMode.{0} batch:{1}")
732+
@MethodSource("testInOrderAckPauseUntilAckedParamters")
746733
@SuppressWarnings("unchecked")
747-
private void testInOrderAckPauseUntilAcked(AckMode ackMode, boolean batch) throws Exception {
734+
void testInOrderAckPauseUntilAcked(AckMode ackMode, boolean batch) throws Exception {
748735
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
749736
Consumer<Integer, String> consumer = mock(Consumer.class);
750737
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
@@ -998,18 +985,10 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
998985
verify(consumer, never()).wakeup();
999986
}
1000987

1001-
@Test
1002-
public void testRecordAckMockForeignThread() throws Exception {
1003-
testRecordAckMockForeignThreadGuts(AckMode.MANUAL);
1004-
}
1005-
1006-
@Test
1007-
public void testRecordAckMockForeignThreadImmediate() throws Exception {
1008-
testRecordAckMockForeignThreadGuts(AckMode.MANUAL_IMMEDIATE);
1009-
}
1010-
988+
@ParameterizedTest(name = "{index} AckMode.{0}")
989+
@EnumSource(value = AckMode.class, names = { "MANUAL", "MANUAL_IMMEDIATE" })
1011990
@SuppressWarnings("unchecked")
1012-
private void testRecordAckMockForeignThreadGuts(AckMode ackMode) throws Exception {
991+
void testRecordAckMockForeignThread(AckMode ackMode) throws Exception {
1013992
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
1014993
Consumer<Integer, String> consumer = mock(Consumer.class);
1015994
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
@@ -1499,25 +1478,6 @@ public void onMessage(List<ConsumerRecord<Integer, String>> data) {
14991478
container.stop();
15001479
}
15011480

1502-
@Test
1503-
public void testSeek() throws Exception {
1504-
Map<String, Object> props = KafkaTestUtils.consumerProps("test11", "false", embeddedKafka);
1505-
testSeekGuts(props, topic11, false);
1506-
}
1507-
1508-
@Test
1509-
public void testSeekAutoCommit() throws Exception {
1510-
Map<String, Object> props = KafkaTestUtils.consumerProps("test12", "true", embeddedKafka);
1511-
testSeekGuts(props, topic12, true);
1512-
}
1513-
1514-
@Test
1515-
public void testSeekAutoCommitDefault() throws Exception {
1516-
Map<String, Object> props = KafkaTestUtils.consumerProps("test15", "true", embeddedKafka);
1517-
props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); // test false by default
1518-
testSeekGuts(props, topic15, false);
1519-
}
1520-
15211481
@Test
15221482
public void testSeekBatch() throws Exception {
15231483
logger.info("Start seek batch seek");
@@ -1573,7 +1533,18 @@ public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekC
15731533
container.stop();
15741534
}
15751535

1576-
private void testSeekGuts(Map<String, Object> props, String topic, boolean autoCommit) throws Exception {
1536+
private static Stream<Arguments> testSeekParameters() {
1537+
Map<String, Object> noAutoCommit = KafkaTestUtils.consumerProps("test15", "true", embeddedKafka);
1538+
noAutoCommit.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); // test false by default
1539+
return Stream.of(
1540+
Arguments.of(KafkaTestUtils.consumerProps("test11", "false", embeddedKafka), topic11, false),
1541+
Arguments.of(KafkaTestUtils.consumerProps("test12", "true", embeddedKafka), topic12, true),
1542+
Arguments.of(noAutoCommit, topic15, false));
1543+
}
1544+
1545+
@ParameterizedTest(name = "topic:{1} autocommit:{2}")
1546+
@MethodSource("testSeekParameters")
1547+
void testSeek(Map<String, Object> props, String topic, boolean autoCommit) throws Exception {
15771548
logger.info("Start seek " + topic);
15781549
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
15791550
ContainerProperties containerProps = new ContainerProperties(topic);

‎spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordTests.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,7 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
248248

249249
@Override
250250
@Nullable
251+
@SuppressWarnings("rawtypes")
251252
public ConsumerRecord intercept(ConsumerRecord record, Consumer consumer) {
252253
return new ConsumerRecord(record.topic(), record.partition(), record.offset(), 0L,
253254
TimestampType.NO_TIMESTAMP_TYPE, 0, 0, record.key(), record.value(), record.headers(),

0 commit comments

Comments
 (0)