From e03de25ec44e77d79da55f4c3456b249a2657802 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 3 Dec 2019 11:53:50 -0500 Subject: [PATCH] GH-1323: s-k-test remove hard AssertJ dependency Resolves https://github.com/spring-projects/spring-kafka/issues/1323 * Polish javadocs. --- .../kafka/test/EmbeddedKafkaBroker.java | 19 +++++---- .../kafka/test/utils/ContainerTestUtils.java | 40 +++++++++---------- .../kafka/test/utils/KafkaTestUtils.java | 21 ++++++---- .../kafka/test/utils/KafkaTestUtilsTests.java | 3 +- 4 files changed, 41 insertions(+), 42 deletions(-) diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java index b1116ef9..ed5107a4 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java @@ -16,8 +16,6 @@ package org.springframework.kafka.test; -import static org.assertj.core.api.Assertions.assertThat; - import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; @@ -546,11 +544,12 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { * @param topicsToConsume the topics. */ public void consumeFromEmbeddedTopics(Consumer consumer, String... topicsToConsume) { - HashSet diff = new HashSet<>(Arrays.asList(topicsToConsume)); - diff.removeAll(new HashSet<>(this.topics)); - assertThat(this.topics) - .as("topic(s):'" + diff + "' are not in embedded topic list") - .containsAll(new HashSet<>(Arrays.asList(topicsToConsume))); + List notEmbedded = Arrays.stream(topicsToConsume) + .filter(topic -> !this.topics.contains(topic)) + .collect(Collectors.toList()); + if (notEmbedded.size() > 0) { + throw new IllegalStateException("topic(s):'" + notEmbedded + "' are not in embedded topic list"); + } final AtomicBoolean assigned = new AtomicBoolean(); consumer.subscribe(Arrays.asList(topicsToConsume), new ConsumerRebalanceListener() { @@ -580,9 +579,9 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { .collect(Collectors.toList())); consumer.seekToBeginning(records.partitions()); } - assertThat(assigned.get()) - .as("Failed to be assigned partitions from the embedded topics") - .isTrue(); + if (!assigned.get()) { + throw new IllegalStateException("Failed to be assigned partitions from the embedded topics"); + } logger.debug("Subscription Initiated"); } diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/ContainerTestUtils.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/ContainerTestUtils.java index 4aed3aca..c3c07166 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/ContainerTestUtils.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/ContainerTestUtils.java @@ -16,8 +16,6 @@ package org.springframework.kafka.test.utils; -import static org.assertj.core.api.Assertions.assertThat; - import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Collection; @@ -25,8 +23,6 @@ import java.util.List; import java.util.concurrent.atomic.AtomicReference; import org.springframework.util.ReflectionUtils; -import org.springframework.util.ReflectionUtils.MethodCallback; -import org.springframework.util.ReflectionUtils.MethodFilter; /** * Utilities for testing listener containers. No hard references to container @@ -45,6 +41,10 @@ public final class ContainerTestUtils { * Wait until the container has the required number of assigned partitions. * @param container the container. * @param partitions the number of partitions. + * @throws IllegalStateException if the operation cannot be completed (since 2.3.4) as + * expected. + * @throws ContainerTestUtilsException if the call to the container's + * getAssignedPartitions() method fails. */ public static void waitForAssignment(Object container, int partitions) { if (container.getClass().getSimpleName().contains("KafkaMessageListenerContainer")) { @@ -76,12 +76,14 @@ public final class ContainerTestUtils { try { Thread.sleep(100); // NOSONAR magic # } - catch (InterruptedException e) { + catch (@SuppressWarnings("unused") InterruptedException e) { Thread.currentThread().interrupt(); } } } - assertThat(count).isEqualTo(partitions); + if (count != partitions) { + throw new IllegalStateException(String.format("Expected %d but got %d partitions", partitions, count)); + } } private static void waitForSingleContainerAssignment(Object container, int partitions) { @@ -104,30 +106,24 @@ public final class ContainerTestUtils { try { Thread.sleep(100); // NOSONAR magic # } - catch (InterruptedException e) { + catch (@SuppressWarnings("unused") InterruptedException e) { Thread.currentThread().interrupt(); } } } - assertThat(count).isEqualTo(partitions); + if (count != partitions) { + throw new IllegalStateException(String.format("Expected %d but got %d partitions", partitions, count)); + } } private static Method getAssignedPartitionsMethod(Class clazz) { final AtomicReference theMethod = new AtomicReference(); - ReflectionUtils.doWithMethods(clazz, new MethodCallback() { - - @Override - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - theMethod.set(method); - } - }, new MethodFilter() { - - @Override - public boolean matches(Method method) { - return method.getName().equals("getAssignedPartitions") && method.getParameterTypes().length == 0; - } - }); - assertThat(theMethod.get()).isNotNull(); + ReflectionUtils.doWithMethods(clazz, + method -> theMethod.set(method), + method -> method.getName().equals("getAssignedPartitions") && method.getParameterTypes().length == 0); + if (theMethod.get() == null) { + throw new IllegalStateException(clazz + " has no getAssignedParitions() method"); + } return theMethod.get(); } diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java index 92403253..88202912 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java @@ -16,8 +16,6 @@ package org.springframework.kafka.test.utils; -import static org.assertj.core.api.Assertions.assertThat; - import java.time.Duration; import java.util.Collections; import java.util.HashMap; @@ -131,7 +129,7 @@ public final class KafkaTestUtils { * @param the key type. * @param the value type. * @return the record. - * @throws org.junit.ComparisonFailure if exactly one record is not received. + * @throws IllegalStateException if exactly one record is not received. * @see #getSingleRecord(Consumer, String, long) */ public static ConsumerRecord getSingleRecord(Consumer consumer, String topic) { @@ -146,7 +144,7 @@ public final class KafkaTestUtils { * @param the key type. * @param the value type. * @return the record. - * @throws org.junit.ComparisonFailure if exactly one record is not received. + * @throws IllegalStateException if exactly one record is not received. * @since 2.0 */ public static ConsumerRecord getSingleRecord(Consumer consumer, String topic, long timeout) { @@ -167,15 +165,19 @@ public final class KafkaTestUtils { try { Thread.sleep(50); // NOSONAR magic# } - catch (InterruptedException e) { + catch (@SuppressWarnings("unused") InterruptedException e) { Thread.currentThread().interrupt(); } remaining = expire - System.currentTimeMillis(); } while (!iterator.hasNext() && remaining > 0); - assertThat(iterator.hasNext()).as("No records found for topic").isTrue(); + if (!iterator.hasNext()) { + throw new IllegalStateException("No records found for topic"); + } iterator.next(); - assertThat(iterator.hasNext()).as("More than one record for topic found").isFalse(); + if (iterator.hasNext()) { + throw new IllegalStateException("More than one record for topic found"); + } return received.records(topic).iterator().next(); } @@ -256,6 +258,7 @@ public final class KafkaTestUtils { * @param the key type. * @param the value type. * @return the records. + * @throws IllegalStateException if the poll returns null (since 2.3.4). * @since 2.0 */ public static ConsumerRecords getRecords(Consumer consumer, long timeout) { @@ -267,7 +270,9 @@ public final class KafkaTestUtils { // map to same format as send metadata toString() .map(r -> r.topic() + "-" + r.partition() + "@" + r.offset()) .collect(Collectors.toList())); - assertThat(received).as("null received from consumer.poll()").isNotNull(); + if (received == null) { + throw new IllegalStateException("null received from consumer.poll()"); + } return received; } diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java index 4281ff20..f325547b 100644 --- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java @@ -27,7 +27,6 @@ import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.jupiter.api.Test; -import org.opentest4j.AssertionFailedError; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.context.EmbeddedKafka; @@ -66,7 +65,7 @@ public class KafkaTestUtilsTests { KafkaConsumer consumer = new KafkaConsumer<>(consumerProps); broker.consumeFromEmbeddedTopics(consumer, "singleTopic4", "singleTopic5"); long t1 = System.currentTimeMillis(); - assertThatExceptionOfType(AssertionFailedError.class).isThrownBy(() -> + assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> KafkaTestUtils.getSingleRecord(consumer, "singleTopic5", 2000L)); assertThat(System.currentTimeMillis() - t1).isGreaterThanOrEqualTo(2000L); producer.send(new ProducerRecord<>("singleTopic5", 1, "foo"));