GH-29: Migrate to AssertJ. Enable its Check Rule

Fixes GH-29 (https://github.com/spring-projects/spring-kafka/issues/29)

Polishing

build.gradle

@@ -65,10 +65,11 @@ subprojects { subproject ->
 	targetCompatibility = 1.7
 
 	ext {
+		assertjVersion = '3.3.0'
 		avroVersion = '1.7.6'
 		gsCollectionsVersion = '5.0.0'
 		hamcrestVersion = '1.3'
-		junitVersion = '4.11'
+		junitVersion = '4.12'
 		kafkaVersion = '0.9.0.1'
 		log4jVersion = '1.2.17'
 		mockitoVersion = '1.9.5'
@@ -146,6 +147,7 @@ project ('spring-kafka') {
 		compile "org.apache.kafka:kafka-clients:$kafkaVersion"
 
 		testCompile project (":spring-kafka-test")
+		testCompile "org.assertj:assertj-core:$assertjVersion"
 	}
 }
 
@@ -167,8 +169,8 @@ project ('spring-kafka-test') {
 		compile ("org.mockito:mockito-core:$mockitoVersion") {
 			exclude group: 'org.hamcrest'
 		}
-		compile "org.hamcrest:hamcrest-all:$hamcrestVersion"
-
+		compile ("org.hamcrest:hamcrest-all:$hamcrestVersion", optional)
+		compile ("org.assertj:assertj-core:$assertjVersion", optional)
 	}
 }
 

spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.test.assertj;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.assertj.core.api.Condition;
+
+/**
+ * The AssertJ custom {@link Condition}s.
+ *
+ * @author Artem Bilan
+ */
+public final class KafkaConditions {
+
+	private KafkaConditions() {
+		// private ctor
+	}
+
+	/**
+	 * @param key the key
+	 * @param <K> the type.
+	 * @return a Condition that matches the key in a consumer record.
+	 */
+	public static <K> Condition<ConsumerRecord<K, ?>> key(K key) {
+		return new ConsumerRecordKeyCondition<>(key);
+	}
+
+	/**
+	 * @param value the value.
+	 * @param <V> the type.
+	 * @return a Condition that matches the value in a consumer record.
+	 */
+	public static <V> Condition<ConsumerRecord<?, V>> value(V value) {
+		return new ConsumerRecordValueCondition<>(value);
+	}
+
+	/**
+	 * @param partition the partition.
+	 * @return a Condition that matches the partition in a consumer record.
+	 */
+	public static Condition<ConsumerRecord<?, ?>> partition(int partition) {
+		return new ConsumerRecordPartitionCondition(partition);
+	}
+
+
+	public static class ConsumerRecordKeyCondition<K> extends Condition<ConsumerRecord<K, ?>> {
+
+		private final K key;
+
+		public ConsumerRecordKeyCondition(K key) {
+			super("a ConsumerRecord with 'key' " + key);
+			this.key = key;
+		}
+
+		@Override
+		public boolean matches(ConsumerRecord<K, ?> value) {
+			return value != null && value.key().equals(this.key);
+		}
+
+	}
+
+	public static class ConsumerRecordValueCondition<V> extends Condition<ConsumerRecord<?, V>> {
+
+		private final V payload;
+
+		public ConsumerRecordValueCondition(V payload) {
+			super("a ConsumerRecord with 'value' " + payload);
+			this.payload = payload;
+		}
+
+		@Override
+		public boolean matches(ConsumerRecord<?, V> value) {
+			return value != null && value.value().equals(this.payload);
+		}
+
+	}
+
+	public static class ConsumerRecordPartitionCondition extends Condition<ConsumerRecord<?, ?>> {
+
+		private final int partition;
+
+		public ConsumerRecordPartitionCondition(int partition) {
+			super("a ConsumerRecord with 'partition' " + partition);
+			this.partition = partition;
+		}
+
+		@Override
+		public boolean matches(ConsumerRecord<?, ?> value) {
+			return value != null && value.partition() == this.partition;
+		}
+
+	}
+
+}

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -16,9 +16,7 @@
 
 package org.springframework.kafka.annotation;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
@@ -56,6 +54,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  */
 @ContextConfiguration
@@ -81,26 +80,26 @@ public class EnableKafkaIntegrationTests {
 		waitListening("foo");
 		template.convertAndSend("annotated1", 0, "foo");
 		template.flush();
-		assertTrue(this.listener.latch1.await(10, TimeUnit.SECONDS));
+		assertThat(this.listener.latch1.await(10, TimeUnit.SECONDS)).isTrue();
 
 		waitListening("bar");
 		template.convertAndSend("annotated2", 0, "foo");
 		template.flush();
-		assertTrue(this.listener.latch2.await(10, TimeUnit.SECONDS));
-		assertNotNull(this.listener.partition);
+		assertThat(this.listener.latch2.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.partition).isNotNull();
 
 		waitListening("baz");
 		template.convertAndSend("annotated3", 0, "foo");
 		template.flush();
-		assertTrue(this.listener.latch3.await(10, TimeUnit.SECONDS));
-		assertEquals("foo", this.listener.record.value());
+		assertThat(this.listener.latch3.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.record.value()).isEqualTo("foo");
 
 		waitListening("qux");
 		template.convertAndSend("annotated4", 0, "foo");
 		template.flush();
-		assertTrue(this.listener.latch4.await(10, TimeUnit.SECONDS));
-		assertEquals("foo", this.listener.record.value());
-		assertNotNull(this.listener.ack);
+		assertThat(this.listener.latch4.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.record.value()).isEqualTo("foo");
+		assertThat(this.listener.ack).isNotNull();
 
 		waitListening("fiz");
 		template.convertAndSend("annotated5", 0, 0, "foo");
@@ -108,7 +107,7 @@ public class EnableKafkaIntegrationTests {
 		template.convertAndSend("annotated6", 0, 0, "baz");
 		template.convertAndSend("annotated6", 1, 0, "qux");
 		template.flush();
-		assertTrue(this.listener.latch5.await(10, TimeUnit.SECONDS));
+		assertThat(this.listener.latch5.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
 	private void waitListening(String id) throws InterruptedException {
@@ -120,7 +119,7 @@ public class EnableKafkaIntegrationTests {
 		while (n++ < 6000 && (kmlc.getAssignedPartitions() == null || kmlc.getAssignedPartitions().size() == 0)) {
 			Thread.sleep(100);
 		}
-		assertTrue(kmlc.getAssignedPartitions().size() > 0);
+		assertThat(kmlc.getAssignedPartitions().size()).isGreaterThan(0);
 	}
 
 	@Configuration

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -16,10 +16,10 @@
 
 package org.springframework.kafka.core;
 
-import static org.junit.Assert.assertThat;
-import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasKey;
-import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasPartition;
-import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasValue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.kafka.test.assertj.KafkaConditions.key;
+import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
+import static org.springframework.kafka.test.assertj.KafkaConditions.value;
 
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
@@ -39,6 +39,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  */
 public class KafkaTemplateTests {
@@ -73,17 +74,17 @@ public class KafkaTemplateTests {
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(TEMPLATE_TOPIC);
 		template.syncConvertAndSend("foo");
-		assertThat(records.poll(10, TimeUnit.SECONDS), hasValue("foo"));
+		assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo"));
 		template.syncConvertAndSend(0, 2, "bar");
 		ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
-		assertThat(received, hasKey(2));
-		assertThat(received, hasPartition(0));
-		assertThat(received, hasValue("bar"));
+		assertThat(received).has(key(2));
+		assertThat(received).has(partition(0));
+		assertThat(received).has(value("bar"));
 		template.syncConvertAndSend(TEMPLATE_TOPIC, 0, 2, "baz");
 		received = records.poll(10, TimeUnit.SECONDS);
-		assertThat(received, hasKey(2));
-		assertThat(received, hasPartition(0));
-		assertThat(received, hasValue("baz"));
+		assertThat(received).has(key(2));
+		assertThat(received).has(partition(0));
+		assertThat(received).has(value("baz"));
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java

@@ -16,11 +16,7 @@
 
 package org.springframework.kafka.listener;
 
-import static org.hamcrest.Matchers.anyOf;
-import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
@@ -110,7 +106,7 @@ public class ConcurrentMessageListenerContainerTests {
 		template.convertAndSend(0, "baz");
 		template.convertAndSend(2, "qux");
 		template.flush();
-		assertTrue(latch.await(60, TimeUnit.SECONDS));
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		container.stop();
 		logger.info("Stop auto");
 	}
@@ -144,7 +140,7 @@ public class ConcurrentMessageListenerContainerTests {
 		template.convertAndSend(0, "baz");
 		template.convertAndSend(2, "qux");
 		template.flush();
-		assertTrue(latch.await(60, TimeUnit.SECONDS));
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		container.stop();
 		logger.info("Stop manual");
 	}
@@ -209,7 +205,7 @@ public class ConcurrentMessageListenerContainerTests {
 		container2.setBeanName("b2");
 		container2.start();
 
-		assertTrue(initialConsumersLatch.await(20, TimeUnit.SECONDS));
+		assertThat(initialConsumersLatch.await(20, TimeUnit.SECONDS)).isTrue();
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
@@ -221,8 +217,8 @@ public class ConcurrentMessageListenerContainerTests {
 		template.convertAndSend(2, "qux");
 		template.flush();
 
-		assertTrue(latch1.await(60, TimeUnit.SECONDS));
-		assertTrue(latch2.await(60, TimeUnit.SECONDS));
+		assertThat(latch1.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(latch2.await(60, TimeUnit.SECONDS)).isTrue();
 		container1.stop();
 		container2.stop();
 
@@ -242,9 +238,9 @@ public class ConcurrentMessageListenerContainerTests {
 			}
 		});
 		resettingContainer.start();
-		assertTrue(latch3.await(60, TimeUnit.SECONDS));
+		assertThat(latch3.await(60, TimeUnit.SECONDS)).isTrue();
 		resettingContainer.stop();
-		assertThat(latch3.getCount(), equalTo(0L));
+		assertThat(latch3.getCount()).isEqualTo(0L);
 
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		cf = new DefaultKafkaConsumerFactory<>(props);
@@ -264,10 +260,10 @@ public class ConcurrentMessageListenerContainerTests {
 			}
 		});
 		resettingContainer.start();
-		assertTrue(latch4.await(60, TimeUnit.SECONDS));
+		assertThat(latch4.await(60, TimeUnit.SECONDS)).isTrue();
 		resettingContainer.stop();
-		assertThat(receivedMessage.get(), anyOf(equalTo("baz"), equalTo("qux")));
-		assertThat(latch4.getCount(), equalTo(0L));
+		assertThat(receivedMessage.get()).isIn("baz", "qux");
+		assertThat(latch4.getCount()).isEqualTo(0L);
 
 		logger.info("Stop auto parts");
 	}
@@ -312,7 +308,7 @@ public class ConcurrentMessageListenerContainerTests {
 		template.convertAndSend(0, "baz");
 		template.convertAndSend(2, "qux");
 		template.flush();
-		assertTrue(latch.await(60, TimeUnit.SECONDS));
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		container.stop();
 		logger.info("Stop " + ackMode);
 	}
@@ -356,10 +352,10 @@ public class ConcurrentMessageListenerContainerTests {
 		List<KafkaMessageListenerContainer<Integer, String>> containers =
 				(List<KafkaMessageListenerContainer<Integer, String>>) new DirectFieldAccessor(
 				container).getPropertyValue("containers");
-		assertEquals(3, containers.size());
+		assertThat(containers.size()).isEqualTo(3);
 		for (int i = 0; i < 3; i++) {
-			assertEquals(i < 2 ? 2 : 3, ((TopicPartition[]) new DirectFieldAccessor(containers.get(i))
-					.getPropertyValue("partitions")).length);
+			assertThat(((TopicPartition[]) new DirectFieldAccessor(containers.get(i))
+					.getPropertyValue("partitions")).length).isEqualTo(i < 2 ? 2 : 3);
 		}
 		container.stop();
 	}
@@ -394,7 +390,7 @@ public class ConcurrentMessageListenerContainerTests {
 		template.convertAndSend(0, "baz");
 		template.convertAndSend(2, "qux");
 		template.flush();
-		assertTrue(latch.await(60, TimeUnit.SECONDS));
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		container.stop();
 		logger.info("Stop exception");
 

spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerTestUtils.java

@@ -16,8 +16,7 @@
 
 package org.springframework.kafka.listener;
 
-import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.List;
 
@@ -47,7 +46,7 @@ public final class ContainerTestUtils {
 				Thread.sleep(100);
 			}
 		}
-		assertThat(count, equalTo(partitions));
+		assertThat(count).isEqualTo(partitions);
 	}
 
 	public static void waitForAssignment(KafkaMessageListenerContainer<Integer, String> container, int partitions)
@@ -63,7 +62,7 @@ public final class ContainerTestUtils {
 				Thread.sleep(100);
 			}
 		}
-		assertThat(count, equalTo(partitions));
+		assertThat(count).isEqualTo(partitions);
 	}
 
 }

src/checkstyle/checkstyle.xml

@@ -81,16 +81,10 @@
 				org.junit.internal.matchers.ThrowableMessageMatcher.*,
 				org.hamcrest.CoreMatchers.*,
 				org.hamcrest.Matchers.*,
-				org.springframework.boot.configurationprocessor.ConfigurationMetadataMatchers.*,
-				org.springframework.boot.configurationprocessor.TestCompiler.*,
 				org.mockito.Mockito.*,
 				org.mockito.Matchers.*,
-				org.springframework.test.web.servlet.request.MockMvcRequestBuilders.*,
-				org.springframework.test.web.servlet.result.MockMvcResultMatchers.*,
-				org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.*,
-				org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.*,
-				org.springframework.hateoas.mvc.ControllerLinkBuilder.linkTo,
-				org.springframework.kafka.test.hamcrest.KafkaMatchers.*" />
+				org.springframework.kafka.test.hamcrest.KafkaMatchers.*,
+				org.springframework.kafka.test.assertj.KafkaConditions.*" />
 		</module>
 		<module name="IllegalImport" />
 		<module name="RedundantImport" />
@@ -106,10 +100,10 @@
 		</module>
 
 		<!-- Javadoc Comments -->
-		<!--<module name="JavadocType">-->
-			<!--<property name="scope" value="package"/>-->
-			<!--<property name="authorFormat" value=".+\s.+"/>-->
-		<!--</module>-->
+<!--		<module name="JavadocType"> -->
+<!--			<property name="scope" value="package"/> -->
+<!--			<property name="authorFormat" value=".+\s.+"/> -->
+<!--		</module> -->
 <!--		<module name="JavadocMethod"> -->
 <!--			<property name="allowMissingJavadoc" value="true" /> -->
 <!--		</module> -->
@@ -157,13 +151,13 @@
 <!-- 				value="Please use BDDMockito imports." /> -->
 <!-- 			<property name="ignoreComments" value="true" /> -->
 <!-- 		</module> -->
-		<!--<module name="RegexpSinglelineJava">-->
-			<!--<property name="maximum" value="0"/>-->
-			<!--<property name="format" value="org\.junit\.Assert\.assert" />-->
-			<!--<property name="message"-->
-				<!--value="Please use AssertJ imports." />-->
-			<!--<property name="ignoreComments" value="true" />-->
-		<!--</module>-->
+		<module name="RegexpSinglelineJava">
+			<property name="maximum" value="0"/>
+			<property name="format" value="org\.junit\.Assert\.assert" />
+			<property name="message"
+				value="Please use AssertJ imports." />
+			<property name="ignoreComments" value="true" />
+		</module>
 		<module name="Regexp">
 			<property name="format" value="[ \t]+$" />
 			<property name="illegalPattern" value="true" />

src/reference/asciidoc/testing.adoc

@@ -60,12 +60,14 @@ The `o.s.kafka.test.hamcrest.KafkaMatchers` provides the following matchers:
 ----
 /**
  * @param key the key
+ * @param <K> the type.
  * @return a Matcher that matches the key in a consumer record.
  */
 public static <K> Matcher<ConsumerRecord<K, ?>> hasKey(K key) { ... }
 
 /**
  * @param value the value.
+ * @param <V> the type.
  * @return a Matcher that matches the value in a consumer record.
  */
 public static <V> Matcher<ConsumerRecord<?, V>> hasValue(V value) { ... }
@@ -77,6 +79,31 @@ public static <V> Matcher<ConsumerRecord<?, V>> hasValue(V value) { ... }
 public static Matcher<ConsumerRecord<?, ?>> hasPartition(int partition) { ... }
 ----
 
+==== AssertJ Conditions
+
+[source, java]
+----
+/**
+ * @param key the key
+ * @param <K> the type.
+ * @return a Condition that matches the key in a consumer record.
+ */
+public static <K> Condition<ConsumerRecord<K, ?>> key(K key) { ... }
+
+/**
+ * @param value the value.
+ * @param <V> the type.
+ * @return a Condition that matches the value in a consumer record.
+ */
+public static <V> Condition<ConsumerRecord<?, V>> value(V value) { ... }
+
+/**
+ * @param partition the partition.
+ * @return a Condition that matches the partition in a consumer record.
+ */
+public static Condition<ConsumerRecord<?, ?>> partition(int partition) { ... }
+----
+
 ==== Example
 
 Putting it all together:
@@ -131,3 +158,23 @@ public class KafkaTemplateTests {
 
 }
 ----
+
+The above uses the hamcrest matchers; with `AssertJ`, the final part looks like this...
+
+[source, java]
+----
+...
+        assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo"));
+        template.syncConvertAndSend(0, 2, "bar");
+        ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
+        assertThat(received).has(key(2));
+        assertThat(received).has(partition(0));
+        assertThat(received).has(value("bar"));
+        template.syncConvertAndSend(TEMPLATE_TOPIC, 0, 2, "baz");
+        received = records.poll(10, TimeUnit.SECONDS);
+        assertThat(received).has(key(2));
+        assertThat(received).has(partition(0));
+        assertThat(received).has(value("baz"));
+    }
+}
+----