Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1323: s-k-test remove hard AssertJ dependency

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/1323

* Polish javadocs.
This commit is contained in:
Gary Russell
2019-12-03 11:53:50 -05:00
committed by Artem Bilan
parent 0a95ec9b6e
commit e03de25ec4
4 changed files with 41 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java
View File

@@ -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<String> 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<String> 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");
}

40
spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/ContainerTestUtils.java
View File

@@ -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<Method> theMethod = new AtomicReference<Method>();
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();
}

21
spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java
View File

@@ -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 <K> the key type.
* @param <V> 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 <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consumer, String topic) {
@@ -146,7 +144,7 @@ public final class KafkaTestUtils {
* @param <K> the key type.
* @param <V> 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 <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> 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 <K> the key type.
* @param <V> the value type.
* @return the records.
* @throws IllegalStateException if the poll returns null (since 2.3.4).
* @since 2.0
*/
public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> 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;
}

3
spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java
View File

@@ -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<Integer, String> 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"));