GH-1299: Add AssertJ keyValue Condition

Resolves https://github.com/spring-projects/spring-kafka/issues/1299

**cherry pick to 2.2.x**

* Fix NPE when expected value is null

* Fix since; add `allOf` test

* Add new condition to docs

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

@@ -34,7 +34,7 @@ public final class KafkaConditions {
 	}
 
 	/**
-	 * @param key the key
+	 * @param key the key.
 	 * @param <K> the type.
 	 * @return a Condition that matches the key in a consumer record.
 	 */
@@ -51,6 +51,18 @@ public final class KafkaConditions {
 		return new ConsumerRecordValueCondition<>(value);
 	}
 
+	/**
+	 * @param key the key.
+	 * @param value the value.
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 * @return a Condition that matches the key in a consumer record.
+	 * @since 2.2.12
+	 */
+	public static <K, V> Condition<ConsumerRecord<K, V>> keyValue(K key, V value) {
+		return new ConsumerRecordKeyValueCondition<K, V>(key, value);
+	}
+
 	/**
 	 * @param value the timestamp.
 	 * @return a Condition that matches the timestamp value in a consumer record.
@@ -90,7 +102,8 @@ public final class KafkaConditions {
 
 		@Override
 		public boolean matches(ConsumerRecord<K, ?> value) {
-			return value != null && ((value.key() == null && this.key == null) || value.key().equals(this.key));
+			return value != null && ((value.key() == null && this.key == null)
+					|| (value.key() != null && value.key().equals(this.key)));
 		}
 
 	}
@@ -106,7 +119,28 @@ public final class KafkaConditions {
 
 		@Override
 		public boolean matches(ConsumerRecord<?, V> value) {
-			return value != null && value.value().equals(this.payload);
+			return value != null
+					&& (value.value() == null && this.payload == null
+					|| (value.value() != null && value.value().equals(this.payload)));
+		}
+
+	}
+
+	public static class ConsumerRecordKeyValueCondition<K, V> extends Condition<ConsumerRecord<K, V>> {
+
+		private final ConsumerRecordKeyCondition<K> keyCondition;
+
+		private final ConsumerRecordValueCondition<V> valueCondition;
+
+		public ConsumerRecordKeyValueCondition(K key, V value) {
+			super("a ConsumerRecord with 'key' " + key + " and 'value' " + value);
+			this.keyCondition = new ConsumerRecordKeyCondition<>(key);
+			this.valueCondition = new ConsumerRecordValueCondition<>(value);
+		}
+
+		@Override
+		public boolean matches(ConsumerRecord<K, V> value) {
+			return this.keyCondition.matches(value) && this.valueCondition.matches(value);
 		}
 
 	}

spring-kafka-test/src/test/java/org/springframework/kafka/test/assertj/KafkaConditionsTests.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 static org.assertj.core.api.Assertions.allOf;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.kafka.test.assertj.KafkaConditions.keyValue;
+import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Gary Russell
+ * @since 2.2.12
+ *
+ */
+public class KafkaConditionsTests {
+
+	@Test
+	void testKeyValue() {
+		ConsumerRecord<Integer, String> record = new ConsumerRecord<>("topic", 42, 0, 23, "foo");
+		assertThat(record).has(allOf(keyValue(23, "foo"), partition(42)));
+		record = new ConsumerRecord<>("topic", 42, 0, 23, null);
+		assertThat(record).has(keyValue(23, null));
+		record = new ConsumerRecord<>("topic", 42, 0, null, "foo");
+		assertThat(record).has(keyValue(null, "foo"));
+		record = new ConsumerRecord<>("topic", 42, 0, null, null);
+		assertThat(record).has(keyValue(null, null));
+		assertThat(record).doesNotHave(keyValue(23, null));
+		assertThat(record).doesNotHave(keyValue(null, "foo"));
+		record = null;
+		assertThat(record).doesNotHave(keyValue(null, null));
+	}
+
+}

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

@@ -16,9 +16,11 @@
 
 package org.springframework.kafka.core;
 
+import static org.assertj.core.api.Assertions.allOf;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.springframework.kafka.test.assertj.KafkaConditions.key;
+import static org.springframework.kafka.test.assertj.KafkaConditions.keyValue;
 import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
 import static org.springframework.kafka.test.assertj.KafkaConditions.timestamp;
 import static org.springframework.kafka.test.assertj.KafkaConditions.value;
@@ -114,21 +116,15 @@ public class KafkaTemplateTests {
 
 		template.sendDefault(0, 2, "bar");
 		ConsumerRecord<Integer, String> received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
-		assertThat(received).has(key(2));
-		assertThat(received).has(partition(0));
-		assertThat(received).has(value("bar"));
+		assertThat(received).has(allOf(keyValue(2, "bar"), partition(0)));
 
 		template.send(INT_KEY_TOPIC, 0, 2, "baz");
 		received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
-		assertThat(received).has(key(2));
-		assertThat(received).has(partition(0));
-		assertThat(received).has(value("baz"));
+		assertThat(received).has(allOf(keyValue(2, "baz"), partition(0)));
 
 		template.send(INT_KEY_TOPIC, 0, null, "qux");
 		received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
-		assertThat(received).has(key((Integer) null));
-		assertThat(received).has(partition(0));
-		assertThat(received).has(value("qux"));
+		assertThat(received).has(allOf(keyValue(null, "qux"), partition(0)));
 
 		template.send(MessageBuilder.withPayload("fiz")
 				.setHeader(KafkaHeaders.TOPIC, INT_KEY_TOPIC)
@@ -136,18 +132,14 @@ public class KafkaTemplateTests {
 				.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
 				.build());
 		received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
-		assertThat(received).has(key(2));
-		assertThat(received).has(partition(0));
-		assertThat(received).has(value("fiz"));
+		assertThat(received).has(allOf(keyValue(2, "fiz"), partition(0)));
 
 		template.send(MessageBuilder.withPayload("buz")
 				.setHeader(KafkaHeaders.PARTITION_ID, 0)
 				.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
 				.build());
 		received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
-		assertThat(received).has(key(2));
-		assertThat(received).has(partition(0));
-		assertThat(received).has(value("buz"));
+		assertThat(received).has(allOf(keyValue(2, "buz"), partition(0)));
 
 		Map<MetricName, ? extends Metric> metrics = template.execute(Producer::metrics);
 		assertThat(metrics).isNotNull();

src/reference/asciidoc/testing.adoc

@@ -453,6 +453,16 @@ public static <K> Condition<ConsumerRecord<K, ?>> key(K key) { ... }
  */
 public static <V> Condition<ConsumerRecord<?, V>> value(V value) { ... }
 
+/**
+ * @param key the key.
+ * @param value the value.
+ * @param <K> the key type.
+ * @param <V> the value type.
+ * @return a Condition that matches the key in a consumer record.
+ * @since 2.2.12
+ */
+public static <K, V> Condition<ConsumerRecord<K, V>> keyValue(K key, V value) { ... }
+
 /**
  * @param partition the partition.
  * @return a Condition that matches the partition in a consumer record.
@@ -547,13 +557,13 @@ With `AssertJ`, the final part looks like the following code:
 assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo"));
 template.sendDefault(0, 2, "bar");
 ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
+// using individual assertions
 assertThat(received).has(key(2));
-assertThat(received).has(partition(0));
 assertThat(received).has(value("bar"));
+assertThat(received).has(partition(0));
 template.send(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"));
+// using allOf()
+assertThat(received).has(allOf(keyValue(2, "baz"), partition(0)));
 ----
 ====