Added support for Spring Kafka

fixes gh-877

README.adoc

@@ -198,7 +198,7 @@ image::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/{spring
 
 Go to `File` -> `Settings` -> `Other settings` -> `Checkstyle`. There click on the `+` icon in the `Configuration file` section. There, you'll have to define where the checkstyle rules should be picked from. In the image above, we've picked the rules from the cloned Spring Cloud Build repository. However, you can point to the Spring Cloud Build's GitHub repository (e.g. for the `checkstyle.xml` : `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/main/resources/checkstyle.xml`). We need to provide the following variables:
 
-- `checkstyle.header.file` - please point it to the Spring Cloud Build's, `spring-cloud-build-tools/src/main/resources/checkstyle/checkstyle-header.txt` file either in your cloned repo or via the `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/main/resources/checkstyle-header.txt` URL.
+- `checkstyle.header.file` - please point it to the Spring Cloud Build's, `spring-cloud-build-tools/src/main/resources/checkstyle-header.txt` file either in your cloned repo or via the `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/main/resources/checkstyle-header.txt` URL.
 - `checkstyle.suppressions.file` - default suppressions. Please point it to the Spring Cloud Build's, `spring-cloud-build-tools/src/checkstyle/checkstyle-suppressions.xml` file either in your cloned repo or via the `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/checkstyle/checkstyle-suppressions.xml` URL.
 - `checkstyle.additional.suppressions.file` - this variable corresponds to suppressions in your local project. E.g. you're working on `spring-cloud-contract`. Then point to the `project-root/src/checkstyle/checkstyle-suppressions.xml` folder. Example for `spring-cloud-contract` would be: `/home/username/spring-cloud-contract/src/checkstyle/checkstyle-suppressions.xml`.
 

docs/src/main/asciidoc/_project-features-messaging.adoc

@@ -101,7 +101,8 @@ You can use one of the following four integration configurations:
 * Spring Integration
 * Spring Cloud Stream
 * Spring AMQP
-* Spring JMS
+* Spring JMS (requires embedded broker)
+* Spring Kafka (requires embedded broker)
 
 Since we use Spring Boot, if you have added one of these libraries to the classpath, all
 the messaging configuration is automatically set up.
@@ -1150,4 +1151,150 @@ Since the route is set for you, you can send a message to the `{output_name}` de
 ----
 include::{tests_path}/spring-cloud-contract-stub-runner-jms/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/jms/JmsStubRunnerSpec.groovy[tags=trigger_no_output,indent=0]
 ----
+====
+
+[[features-messaging-stub-runner-kafka]]
+=== Consumer Side Messaging With Spring Kafka
+
+Spring Cloud Contract Stub Runner's messaging module provides an easy way to
+integrate with Spring Kafka.
+
+The integration assumes that you have a running instance of a embedded Kafka broker (via the `spring-kafka-test` dependency).
+
+[[features-messaging-stub-runner-kafka-adding]]
+==== Adding the Runner to the Project
+
+You need to have both Spring Kafka, Spring Kafka Test (to run the `@EmbeddedBroker`) and Spring Cloud Contract Stub Runner on the classpath. Remember to annotate your test class
+with `@AutoConfigureStubRunner`.
+
+With Kafka integration, in order to poll for a single message we need to register a consumer upon Spring context startup. That may lead to a situation that, when you're on the consumer side, Stub Runner can register an additional consumer for the same group id and topic. That could lead to a situation that only one of the components would actually poll for the message. Since on the consumer side you have both the Spring Cloud Contract Stub Runner and Spring Cloud Contract Verifier classpath, we need to be able to switch off such behaviour. That's done automatically via the `stubrunner.kafka.initializer.enabled` flag, that will disable the Contact Verifier consumer registration. If your application is both the consumer and the producer of a kafka message, you might need to manually toggle that property to `false` in the base class of your generated tests.
+
+:input_name: input
+:output_name: output
+
+[[features-messaging-stub-runner-kafka-example]]
+==== Examples
+
+Assume that the stub structure looks as follows:
+
+====
+[source,bash,indent=0]
+----
+├── stubs
+    ├── bookDeleted.groovy
+    ├── bookReturned1.groovy
+    └── bookReturned2.groovy
+
+----
+====
+
+Further assume the following test configuration (notice the `spring.kafka.bootstrap-servers` pointing to the embedded broker's IP via `${spring.embedded.kafka.brokers}`):
+
+====
+[source,yml,indent=0]
+----
+stubrunner:
+  repository-root: stubs:classpath:/stubs/
+  ids: my:stubs
+  stubs-mode: remote
+spring:
+  kafka:
+    bootstrap-servers: ${spring.embedded.kafka.brokers}
+    producer:
+      properties:
+        "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer"
+        "spring.json.trusted.packages": "*"
+    consumer:
+      properties:
+        "value.deserializer": "org.springframework.kafka.support.serializer.JsonDeserializer"
+        "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer"
+        "spring.json.trusted.packages": "*"
+      group-id: groupId
+----
+====
+
+Now consider the following contracts (we number them 1 and 2):
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=sample_dsl,indent=0]
+----
+
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=sample_dsl_2,indent=0]
+----
+====
+
+[[features-messaging-stub-runner-kafka-scenario1]]
+===== Scenario 1 (No Input Message)
+
+To trigger a message from the `return_book_1` label, we use the `StubTigger` interface, as follows:
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_trigger,indent=0]
+----
+====
+
+Next, we want to listen to the output of the message sent to `{output_name}`:
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_trigger_receive,indent=0]
+----
+====
+
+The received message would then pass the following assertions:
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_trigger_message,indent=0]
+----
+====
+
+[[features-messaging-stub-runner-kafka-scenario2]]
+===== Scenario 2 (Output Triggered by Input)
+
+Since the route is set for you, you can send a message to the `{output_name}` destination.
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_send,indent=0]
+----
+====
+
+Next, we want to listen to the output of the message sent to `{output_name}`, as follows:
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_receive,indent=0]
+----
+====
+
+The received message would pass the following assertions:
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_receive_message,indent=0]
+----
+====
+
+[[features-messaging-stub-runner-kafka-scenario3]]
+===== Scenario 3 (Input with No Output)
+
+Since the route is set for you, you can send a message to the `{output_name}` destination, as follows:
+
+====
+[source,groovy]
+----
+include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=trigger_no_output,indent=0]
+----
 ====

spring-cloud-contract-stub-runner/pom.xml

@@ -70,6 +70,16 @@
 			<artifactId>spring-boot-starter-activemq</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.camel</groupId>
 			<artifactId>camel-spring-boot-starter</artifactId>

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaConfiguration.java

@@ -0,0 +1,146 @@
+/*
+ * Copyright 2013-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.cloud.contract.stubrunner.messaging.kafka;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.cloud.contract.spec.Contract;
+import org.springframework.cloud.contract.stubrunner.BatchStubRunner;
+import org.springframework.cloud.contract.stubrunner.StubConfiguration;
+import org.springframework.cloud.contract.verifier.messaging.kafka.ContractVerifierKafkaConfiguration;
+import org.springframework.cloud.contract.verifier.messaging.kafka.KafkaStubMessagesInitializer;
+import org.springframework.cloud.contract.verifier.util.MapConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.GenericMessageListener;
+import org.springframework.kafka.listener.KafkaMessageListenerContainer;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.util.StringUtils;
+
+/**
+ * Spring Integration configuration that iterates over the downloaded Groovy DSLs and
+ * registers a flow for each DSL.
+ *
+ * @author Marcin Grzejszczak
+ */
+@Configuration
+@ConditionalOnClass({ KafkaTemplate.class, EmbeddedKafkaBroker.class })
+@ConditionalOnProperty(name = "stubrunner.kafka.enabled", havingValue = "true",
+		matchIfMissing = true)
+@ConditionalOnBean(EmbeddedKafkaBroker.class)
+@AutoConfigureBefore(ContractVerifierKafkaConfiguration.class)
+public class StubRunnerKafkaConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean
+	@ConditionalOnProperty(name = "stubrunner.kafka.initializer.enabled",
+			havingValue = "true", matchIfMissing = true)
+	KafkaStubMessagesInitializer stubRunnerKafkaStubMessagesInitializer() {
+		return (broker, kafkaProperties) -> new HashMap<>();
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(name = "stubFlowRegistrar")
+	public FlowRegistrar stubFlowRegistrar(ConfigurableListableBeanFactory beanFactory,
+			BatchStubRunner batchStubRunner) {
+		Map<StubConfiguration, Collection<Contract>> contracts = batchStubRunner
+				.getContracts();
+		for (Entry<StubConfiguration, Collection<Contract>> entry : contracts
+				.entrySet()) {
+			StubConfiguration key = entry.getKey();
+			Collection<Contract> value = entry.getValue();
+			String name = key.getGroupId() + "_" + key.getArtifactId();
+			MultiValueMap<String, Contract> map = new LinkedMultiValueMap<>();
+			for (Contract dsl : value) {
+				if (dsl == null) {
+					continue;
+				}
+				if (dsl.getInput() != null && dsl.getInput().getMessageFrom() != null
+						&& StringUtils.hasText(
+								dsl.getInput().getMessageFrom().getClientValue())) {
+					String from = dsl.getInput().getMessageFrom().getClientValue();
+					map.add(from, dsl);
+				}
+			}
+			for (Entry<String, List<Contract>> entries : map.entrySet()) {
+				List<Contract> matchingContracts = entries.getValue();
+				final String flowName = name + "_" + entries.getKey() + "_"
+						+ Math.abs(matchingContracts.hashCode());
+				// listener
+				StubRunnerKafkaRouter router = new StubRunnerKafkaRouter(
+						matchingContracts, beanFactory);
+				StubRunnerKafkaRouter listener = (StubRunnerKafkaRouter) beanFactory
+						.initializeBean(router, flowName);
+				beanFactory.registerSingleton(flowName, listener);
+				registerContainers(beanFactory, matchingContracts, flowName, listener);
+			}
+
+		}
+		return new FlowRegistrar();
+	}
+
+	private void registerContainers(ConfigurableListableBeanFactory beanFactory,
+			List<Contract> matchingContracts, String flowName,
+			StubRunnerKafkaRouter listener) {
+		// listener's container
+		ConsumerFactory consumerFactory = beanFactory.getBean(ConsumerFactory.class);
+		for (Contract matchingContract : matchingContracts) {
+			if (matchingContract.getInput() == null) {
+				continue;
+			}
+			String destination = MapConverter.getStubSideValuesForNonBody(
+					matchingContract.getInput().getMessageFrom()).toString();
+			ContainerProperties containerProperties = new ContainerProperties(
+					destination);
+			KafkaMessageListenerContainer container = listenerContainer(consumerFactory,
+					containerProperties, listener);
+			String containerName = flowName + ".container";
+			Object initializedContainer = beanFactory.initializeBean(container,
+					containerName);
+			beanFactory.registerSingleton(containerName, initializedContainer);
+		}
+	}
+
+	private KafkaMessageListenerContainer listenerContainer(
+			ConsumerFactory consumerFactory, ContainerProperties containerProperties,
+			GenericMessageListener listener) {
+		KafkaMessageListenerContainer container = new KafkaMessageListenerContainer(
+				consumerFactory, containerProperties);
+		container.setupMessageListener(listener);
+		return container;
+	}
+
+	static class FlowRegistrar {
+
+	}
+
+}

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaMessageSelector.java

@@ -0,0 +1,259 @@
+/*
+ * Copyright 2013-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.cloud.contract.stubrunner.messaging.kafka;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.WeakHashMap;
+import java.util.regex.Pattern;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.jayway.jsonpath.DocumentContext;
+import com.jayway.jsonpath.JsonPath;
+import com.toomuchcoding.jsonassert.JsonAssertion;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.contract.spec.Contract;
+import org.springframework.cloud.contract.spec.internal.BodyMatcher;
+import org.springframework.cloud.contract.spec.internal.BodyMatchers;
+import org.springframework.cloud.contract.spec.internal.FromFileProperty;
+import org.springframework.cloud.contract.spec.internal.Header;
+import org.springframework.cloud.contract.spec.internal.RegexProperty;
+import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierObjectMapper;
+import org.springframework.cloud.contract.verifier.util.ContentType;
+import org.springframework.cloud.contract.verifier.util.ContentUtils;
+import org.springframework.cloud.contract.verifier.util.JsonPaths;
+import org.springframework.cloud.contract.verifier.util.JsonToJsonPathsConverter;
+import org.springframework.cloud.contract.verifier.util.MapConverter;
+import org.springframework.cloud.contract.verifier.util.MethodBufferingJsonVerifiable;
+import org.springframework.messaging.Message;
+
+/**
+ * Passes through a message that matches the one defined in the DSL.
+ *
+ * @author Marcin Grzejszczak
+ */
+class StubRunnerKafkaMessageSelector {
+
+	private static final Map<Message<?>, Contract> CACHE = Collections
+			.synchronizedMap(new WeakHashMap<>());
+
+	private static final Log log = LogFactory
+			.getLog(StubRunnerKafkaMessageSelector.class);
+
+	private final List<Contract> groovyDsls;
+
+	private final ContractVerifierObjectMapper objectMapper = new ContractVerifierObjectMapper();
+
+	StubRunnerKafkaMessageSelector(List<Contract> groovyDsls) {
+		this.groovyDsls = groovyDsls;
+	}
+
+	Contract matchingContract(Message<?> message) {
+		if (CACHE.containsKey(message)) {
+			return CACHE.get(message);
+		}
+		Contract contract = getContract(message);
+		if (contract != null) {
+			CACHE.put(message, contract);
+		}
+		return contract;
+	}
+
+	void updateCache(Message<?> message, Contract contract) {
+		CACHE.put(message, contract);
+	}
+
+	private Contract getContract(Message<?> message) {
+		for (Contract groovyDsl : this.groovyDsls) {
+			Contract contract = matchContract(message, groovyDsl);
+			if (contract != null) {
+				return contract;
+			}
+		}
+		return null;
+	}
+
+	private Contract matchContract(Message<?> message, Contract groovyDsl) {
+		List<String> unmatchedHeaders = headersMatch(message, groovyDsl);
+		if (!unmatchedHeaders.isEmpty()) {
+			if (log.isDebugEnabled()) {
+				log.debug("Contract [" + groovyDsl
+						+ "] hasn't matched the following headers " + unmatchedHeaders);
+			}
+			return null;
+		}
+		Object inputMessage = message.getPayload();
+		Object dslBody = MapConverter
+				.getStubSideValues(groovyDsl.getInput().getMessageBody());
+		if (dslBody instanceof FromFileProperty) {
+			if (log.isDebugEnabled()) {
+				log.debug("Will compare file content");
+			}
+			FromFileProperty property = (FromFileProperty) dslBody;
+			if (property.isString()) {
+				// continue processing as if body was pure string
+				dslBody = property.asString();
+			}
+			else if (!(inputMessage instanceof byte[])) {
+				if (log.isDebugEnabled()) {
+					log.debug(
+							"Contract provided byte comparison, but the input message is of type ["
+									+ inputMessage.getClass()
+									+ "]. Can't compare the two.");
+				}
+				return null;
+			}
+			else {
+				boolean matches = Arrays.equals(property.asBytes(),
+						(byte[]) inputMessage);
+				if (log.isDebugEnabled() && !matches) {
+					log.debug(
+							"Contract provided byte comparison, but the byte arrays don't match");
+				}
+				return matches ? groovyDsl : null;
+			}
+		}
+		if (matchViaContent(groovyDsl, inputMessage, dslBody)) {
+			return groovyDsl;
+		}
+		return null;
+	}
+
+	private boolean matchViaContent(Contract groovyDsl, Object inputMessage,
+			Object dslBody) {
+		boolean matches;
+		ContentType type = ContentUtils.getClientContentType(inputMessage,
+				groovyDsl.getInput().getMessageHeaders());
+		if (type == ContentType.JSON) {
+			BodyMatchers matchers = groovyDsl.getInput().getBodyMatchers();
+			matches = matchesForJsonPayload(groovyDsl, inputMessage, matchers, dslBody);
+		}
+		else if (dslBody instanceof RegexProperty && inputMessage instanceof String) {
+			Pattern pattern = ((RegexProperty) dslBody).getPattern();
+			matches = pattern.matcher((String) inputMessage).matches();
+			bodyUnmatchedLog(dslBody, matches, pattern);
+		}
+		else {
+			matches = dslBody.equals(inputMessage);
+			bodyUnmatchedLog(dslBody, matches, inputMessage);
+		}
+		return matches;
+	}
+
+	private void bodyUnmatchedLog(Object dslBody, boolean matches, Object pattern) {
+		if (log.isDebugEnabled() && !matches) {
+			log.debug("Body was supposed to " + unmatchedText(pattern)
+					+ " but the value is [" + dslBody.toString() + "]");
+		}
+	}
+
+	private boolean matchesForJsonPayload(Contract groovyDsl, Object inputMessage,
+			BodyMatchers matchers, Object dslBody) {
+		Object matchingInputMessage = JsonToJsonPathsConverter
+				.removeMatchingJsonPaths(dslBody, matchers);
+		JsonPaths jsonPaths = JsonToJsonPathsConverter
+				.transformToJsonPathWithStubsSideValuesAndNoArraySizeCheck(
+						matchingInputMessage);
+		DocumentContext parsedJson;
+		try {
+			parsedJson = JsonPath
+					.parse(this.objectMapper.writeValueAsString(inputMessage));
+		}
+		catch (JsonProcessingException e) {
+			throw new IllegalStateException("Cannot serialize to JSON", e);
+		}
+		List<String> unmatchedJsonPath = new ArrayList<>();
+		boolean matches = true;
+		for (MethodBufferingJsonVerifiable path : jsonPaths) {
+			matches &= matchesJsonPath(unmatchedJsonPath, parsedJson, path.jsonPath());
+		}
+		if (matchers != null && matchers.hasMatchers()) {
+			for (BodyMatcher matcher : matchers.matchers()) {
+				String jsonPath = JsonToJsonPathsConverter
+						.convertJsonPathAndRegexToAJsonPath(matcher, dslBody);
+				matches &= matchesJsonPath(unmatchedJsonPath, parsedJson, jsonPath);
+			}
+		}
+		if (!unmatchedJsonPath.isEmpty()) {
+			if (log.isDebugEnabled()) {
+				log.debug("Contract [" + groovyDsl + "] didn't match the body due to "
+						+ unmatchedJsonPath);
+			}
+		}
+		return matches;
+	}
+
+	private boolean matchesJsonPath(List<String> unmatchedJsonPath,
+			DocumentContext parsedJson, String jsonPath) {
+		try {
+			JsonAssertion.assertThat(parsedJson).matchesJsonPath(jsonPath);
+			return true;
+		}
+		catch (Exception e) {
+			unmatchedJsonPath.add(e.getLocalizedMessage());
+			return false;
+		}
+	}
+
+	private List<String> headersMatch(Message message, Contract groovyDsl) {
+		List<String> unmatchedHeaders = new ArrayList<>();
+		Map<String, Object> headers = message.getHeaders();
+		for (Header it : groovyDsl.getInput().getMessageHeaders().getEntries()) {
+			String name = it.getName();
+			Object value = it.getClientValue();
+			Object valueInHeader = headers.get(name);
+			valueInHeader = valueInHeader instanceof byte[]
+					? fromByte((byte[]) valueInHeader) : valueInHeader;
+			boolean matches;
+			if (value instanceof RegexProperty) {
+				Pattern pattern = ((RegexProperty) value).getPattern();
+				matches = pattern.matcher(valueInHeader.toString()).matches();
+			}
+			else {
+				matches = valueInHeader != null
+						&& valueInHeader.toString().equals(value.toString());
+			}
+			if (!matches) {
+				unmatchedHeaders.add("Header with name [" + name + "] was supposed to "
+						+ unmatchedText(value) + " but the value is ["
+						+ (valueInHeader != null ? valueInHeader.toString() : "null")
+						+ "]");
+			}
+		}
+		return unmatchedHeaders;
+	}
+
+	private String fromByte(byte[] valueInHeader) {
+		String string = new String(valueInHeader);
+		if (string.startsWith("\"") && string.endsWith("\"")) {
+			return string.substring(1, string.length() - 1);
+		}
+		return string;
+	}
+
+	private String unmatchedText(Object expectedValue) {
+		return expectedValue instanceof RegexProperty
+				? "match pattern [" + ((RegexProperty) expectedValue).pattern() + "]"
+				: "be equal to [" + expectedValue + "]";
+	}
+
+}

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaRouter.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2013-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.cloud.contract.stubrunner.messaging.kafka;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.cloud.contract.spec.Contract;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.MessageListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * @author Marcin Grzejszczak
+ */
+class StubRunnerKafkaRouter implements MessageListener<Object, Object> {
+
+	private static final Log log = LogFactory.getLog(StubRunnerKafkaRouter.class);
+
+	private final StubRunnerKafkaMessageSelector selector;
+
+	private final BeanFactory beanFactory;
+
+	private final List<Contract> contracts;
+
+	private KafkaTemplate kafkaTemplate;
+
+	StubRunnerKafkaRouter(List<Contract> groovyDsls, BeanFactory beanFactory) {
+		this.selector = new StubRunnerKafkaMessageSelector(groovyDsls);
+		this.beanFactory = beanFactory;
+		this.contracts = groovyDsls;
+	}
+
+	private KafkaTemplate kafkaTemplate() {
+		if (this.kafkaTemplate == null) {
+			this.kafkaTemplate = this.beanFactory.getBean(KafkaTemplate.class);
+		}
+		return this.kafkaTemplate;
+	}
+
+	@Override
+	public void onMessage(ConsumerRecord<Object, Object> data) {
+		if (log.isDebugEnabled()) {
+			log.debug("Received message [" + data + "]");
+		}
+		Message<?> message = MessageBuilder.createMessage(data.value(),
+				headers(data.headers()));
+		Contract dsl = this.selector.matchingContract(message);
+		if (dsl != null && dsl.getOutputMessage() != null
+				&& dsl.getOutputMessage().getSentTo() != null) {
+			String destination = dsl.getOutputMessage().getSentTo().getClientValue();
+			if (log.isDebugEnabled()) {
+				log.debug(
+						"Found a matching contract with an output message. Will send it to the ["
+								+ destination + "] destination");
+			}
+			Message<?> transform = new StubRunnerKafkaTransformer(this.contracts)
+					.transform(dsl);
+			String defaultTopic = kafkaTemplate().getDefaultTopic();
+			try {
+				kafkaTemplate().setDefaultTopic(destination);
+				kafkaTemplate().send(transform);
+			}
+			finally {
+				kafkaTemplate().setDefaultTopic(defaultTopic);
+			}
+		}
+	}
+
+	private MessageHeaders headers(Headers headers) {
+		Map<String, Object> map = new HashMap<>();
+		for (Header header : headers) {
+			map.put(header.key(), header.value());
+		}
+		return new MessageHeaders(map);
+	}
+
+	@Override
+	public void onMessage(ConsumerRecord<Object, Object> data,
+			Acknowledgment acknowledgment) {
+		onMessage(data);
+	}
+
+	@Override
+	public void onMessage(ConsumerRecord<Object, Object> data, Consumer<?, ?> consumer) {
+		onMessage(data);
+	}
+
+	@Override
+	public void onMessage(ConsumerRecord<Object, Object> data,
+			Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
+		onMessage(data);
+	}
+
+}

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaTransformer.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2013-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.cloud.contract.stubrunner.messaging.kafka;
+
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.cloud.contract.spec.Contract;
+import org.springframework.cloud.contract.spec.internal.FromFileProperty;
+import org.springframework.cloud.contract.verifier.util.BodyExtractor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * Sends forward a message defined in the DSL.
+ *
+ * @author Marcin Grzejszczak
+ */
+class StubRunnerKafkaTransformer {
+
+	private final StubRunnerKafkaMessageSelector selector;
+
+	StubRunnerKafkaTransformer(List<Contract> groovyDsls) {
+		this.selector = new StubRunnerKafkaMessageSelector(groovyDsls);
+	}
+
+	public Message<?> transform(Contract groovyDsl) {
+		Object outputBody = outputBody(groovyDsl);
+		Map<String, Object> headers = groovyDsl.getOutputMessage().getHeaders()
+				.asStubSideMap();
+		Message newMessage = MessageBuilder.createMessage(outputBody,
+				new MessageHeaders(headers));
+		this.selector.updateCache(newMessage, groovyDsl);
+		return newMessage;
+	}
+
+	private Object outputBody(Contract groovyDsl) {
+		Object outputBody = BodyExtractor
+				.extractClientValueFromBody(groovyDsl.getOutputMessage().getBody());
+		if (outputBody instanceof FromFileProperty) {
+			FromFileProperty property = (FromFileProperty) outputBody;
+			return property.asBytes();
+		}
+		return BodyExtractor.extractStubValueFrom(outputBody);
+	}
+
+}

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/provider/wiremock/StubRunnerWireMockTestExecutionListener.java

@@ -46,7 +46,8 @@ public final class StubRunnerWireMockTestExecutionListener
 			}
 			return;
 		}
-		if (WireMockHttpServerStub.SERVERS.values().stream().noneMatch(p -> p.random)) {
+		if (!WireMockHttpServerStub.SERVERS.isEmpty() && WireMockHttpServerStub.SERVERS
+				.values().stream().noneMatch(p -> p.random)) {
 			if (log.isWarnEnabled()) {
 				log.warn("You've used fixed ports for WireMock setup - "
 						+ "will mark context as dirty. Please use random ports, as much "

spring-cloud-contract-stub-runner/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -65,6 +65,24 @@
       "type": "java.lang.Boolean",
       "description": "Whether to enable Stub Runner integration with Spring Cloud Stream.",
       "defaultValue": true
+    },
+    {
+      "name": "stubrunner.jms.enabled",
+      "type": "java.lang.Boolean",
+      "description": "Whether to enable Stub Runner integration with Spring JMS.",
+      "defaultValue": true
+    },
+    {
+      "name": "stubrunner.kafka.enabled",
+      "type": "java.lang.Boolean",
+      "description": "Whether to enable Stub Runner integration with Spring Kafka.",
+      "defaultValue": true
+    },
+    {
+      "name": "stubrunner.kafka.initializer.enabled",
+      "type": "java.lang.Boolean",
+      "description": "Whether to allow Stub Runner to take care of polling for messages instead of the KafkaStubMessages component. The latter should be used only on the producer side.",
+      "defaultValue": true
     }
   ]
 }

spring-cloud-contract-verifier/pom.xml

@@ -36,6 +36,16 @@
 			<artifactId>spring-jms</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>jakarta.jms</groupId>
 			<artifactId>jakarta.jms-api</artifactId>

spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java

@@ -75,14 +75,7 @@ public class SpringAmqpStubMessages implements MessageVerifier<Message> {
 		Assert.isTrue(
 				mockingDetails(rabbitTemplate).isSpy()
 						|| mockingDetails(rabbitTemplate).isMock(),
-				"StubRunner AMQP will work only if RabbiTemplate is a spy"); // we get
-																				// send
-																				// messages
-																				// by
-																				// capturing
-																				// arguments
-																				// on the
-																				// spy
+				"StubRunner AMQP will work only if RabbiTemplate is a spy");
 		this.rabbitTemplate = rabbitTemplate;
 		this.messageListenerAccessor = messageListenerAccessor;
 	}
@@ -95,14 +88,7 @@ public class SpringAmqpStubMessages implements MessageVerifier<Message> {
 		Assert.isTrue(
 				mockingDetails(rabbitTemplate).isSpy()
 						|| mockingDetails(rabbitTemplate).isMock(),
-				"StubRunner AMQP will work only if RabbiTemplate is a spy"); // we get
-																				// send
-																				// messages
-																				// by
-																				// capturing
-																				// arguments
-																				// on the
-																				// spy
+				"StubRunner AMQP will work only if RabbiTemplate is a spy");
 		this.rabbitTemplate = rabbitTemplate;
 		this.messageListenerAccessor = messageListenerAccessor;
 		this.rabbitProperties = rabbitProperties;

spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaConfiguration.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2013-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.cloud.contract.verifier.messaging.kafka;
+
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.cloud.contract.verifier.messaging.MessageVerifier;
+import org.springframework.cloud.contract.verifier.messaging.integration.ContractVerifierIntegrationConfiguration;
+import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessage;
+import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessaging;
+import org.springframework.cloud.contract.verifier.messaging.noop.NoOpContractVerifierAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Marcin Grzejszczak
+ */
+@Configuration
+@ConditionalOnClass({ KafkaTemplate.class, EmbeddedKafkaBroker.class })
+@ConditionalOnProperty(name = "stubrunner.kafka.enabled", havingValue = "true",
+		matchIfMissing = true)
+@AutoConfigureBefore({ ContractVerifierIntegrationConfiguration.class,
+		NoOpContractVerifierAutoConfiguration.class })
+@ConditionalOnBean(EmbeddedKafkaBroker.class)
+public class ContractVerifierKafkaConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean
+	MessageVerifier<Message<?>> contractVerifierKafkaMessageExchange(
+			KafkaTemplate kafkaTemplate, EmbeddedKafkaBroker broker,
+			KafkaProperties kafkaProperties, KafkaStubMessagesInitializer initializer) {
+		return new KafkaStubMessages(kafkaTemplate, broker, kafkaProperties, initializer);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	KafkaStubMessagesInitializer contractVerifierKafkaStubMessagesInitializer() {
+		return new ContractVerifierKafkaStubMessagesInitializer();
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	ContractVerifierMessaging<Message<?>> contractVerifierKafkaMessaging(
+			MessageVerifier<Message<?>> exchange) {
+		return new ContractVerifierKafkaHelper(exchange);
+	}
+
+}
+
+class ContractVerifierKafkaHelper extends ContractVerifierMessaging<Message<?>> {
+
+	ContractVerifierKafkaHelper(MessageVerifier<Message<?>> exchange) {
+		super(exchange);
+	}
+
+	@Override
+	protected ContractVerifierMessage convert(Message<?> message) {
+		return new ContractVerifierMessage(message.getPayload(), message.getHeaders());
+	}
+
+}

spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaStubMessagesInitializer.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2013-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.cloud.contract.verifier.messaging.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+class ContractVerifierKafkaStubMessagesInitializer
+		implements KafkaStubMessagesInitializer {
+
+	private static final Log log = LogFactory
+			.getLog(ContractVerifierKafkaStubMessagesInitializer.class);
+
+	@Override
+	public Map<String, Consumer> initialize(EmbeddedKafkaBroker broker,
+			KafkaProperties kafkaProperties) {
+		Map<String, Consumer> map = new HashMap<>();
+		for (String topic : broker.getTopics()) {
+			map.put(topic, prepareListener(broker, topic, kafkaProperties));
+		}
+		return map;
+	}
+
+	private Consumer prepareListener(EmbeddedKafkaBroker broker, String destination,
+			KafkaProperties kafkaProperties) {
+		Map<String, Object> consumerProperties = KafkaTestUtils.consumerProps(
+				kafkaProperties.getConsumer().getGroupId(), "false", broker);
+		consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(
+				consumerProperties);
+		Consumer<String, String> consumer = consumerFactory.createConsumer();
+		broker.consumeFromAnEmbeddedTopic(consumer, destination);
+		if (log.isDebugEnabled()) {
+			log.debug("Prepared consumer for destination [" + destination + "]");
+		}
+		return consumer;
+	}
+
+}

spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessages.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2013-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.cloud.contract.verifier.messaging.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import net.minidev.json.JSONObject;
+import net.minidev.json.parser.JSONParser;
+import net.minidev.json.parser.ParseException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.cloud.contract.verifier.messaging.MessageVerifier;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+
+class KafkaStubMessages implements MessageVerifier<Message<?>> {
+
+	private static final Log log = LogFactory.getLog(KafkaStubMessages.class);
+
+	private final KafkaTemplate kafkaTemplate;
+
+	private final Receiver receiver;
+
+	KafkaStubMessages(KafkaTemplate kafkaTemplate, EmbeddedKafkaBroker broker,
+			KafkaProperties kafkaProperties, KafkaStubMessagesInitializer initializer) {
+		this.kafkaTemplate = kafkaTemplate;
+		Map<String, Consumer> topicToConsumer = initializer.initialize(broker,
+				kafkaProperties);
+		this.receiver = new Receiver(topicToConsumer);
+	}
+
+	@Override
+	public void send(Message<?> message, String destination) {
+		String defaultTopic = this.kafkaTemplate.getDefaultTopic();
+		try {
+			this.kafkaTemplate.setDefaultTopic(destination);
+			if (log.isDebugEnabled()) {
+				log.debug("Will send a message [" + message + "] to destination ["
+						+ destination + "]");
+			}
+			this.kafkaTemplate.send(message).get(5, TimeUnit.SECONDS);
+			this.kafkaTemplate.flush();
+		}
+		catch (Exception ex) {
+			throw new IllegalStateException(ex);
+		}
+		finally {
+			this.kafkaTemplate.setDefaultTopic(defaultTopic);
+		}
+	}
+
+	@Override
+	public Message receive(String destination, long timeout, TimeUnit timeUnit) {
+		return this.receiver.receive(destination, timeout, timeUnit);
+	}
+
+	@Override
+	public Message receive(String destination) {
+		return receive(destination, 5, TimeUnit.SECONDS);
+	}
+
+	@Override
+	public void send(Object payload, Map headers, String destination) {
+		Message<?> message = MessageBuilder.createMessage(payload,
+				new MessageHeaders(headers));
+		send(message, destination);
+	}
+
+}
+
+class Receiver {
+
+	private final Map<String, Consumer> consumers;
+
+	private static final Log log = LogFactory.getLog(Receiver.class);
+
+	Receiver(Map<String, Consumer> consumers) {
+		this.consumers = consumers;
+	}
+
+	Message receive(String topic, long timeout, TimeUnit timeUnit) {
+		Consumer consumer = this.consumers.get(topic);
+		if (consumer == null) {
+			throw new IllegalStateException(
+					"No consumer set up for topic [" + topic + "]");
+		}
+		ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer,
+				topic, timeUnit.toMillis(timeout));
+		if (log.isDebugEnabled()) {
+			log.debug("Got a single record for destination [" + topic + "]");
+		}
+		return new Record(record).toMessage();
+	}
+
+}
+
+class Record {
+
+	private final ConsumerRecord record;
+
+	Record(ConsumerRecord record) {
+		this.record = record;
+	}
+
+	private Map<String, Object> toMap(Headers headers) {
+		Map<String, Object> map = new HashMap<>();
+		for (Header header : headers) {
+			map.put(header.key(), header.value());
+		}
+		return map;
+	}
+
+	Message toMessage() {
+		Object textPayload = record.value();
+		// sometimes it's a message sometimes just payload
+		MessageHeaders headers = new MessageHeaders(toMap(record.headers()));
+		if (textPayload instanceof String && ((String) textPayload).contains("payload")
+				&& ((String) textPayload).contains("headers")) {
+			try {
+				Object object = new JSONParser(JSONParser.DEFAULT_PERMISSIVE_MODE)
+						.parse((String) textPayload);
+				JSONObject jo = (JSONObject) object;
+				String payload = (String) jo.get("payload");
+				JSONObject headersInJson = (JSONObject) jo.get("headers");
+				Map newHeaders = new HashMap(headers);
+				newHeaders.putAll(headersInJson);
+				return MessageBuilder.createMessage(unquoted(payload),
+						new MessageHeaders(newHeaders));
+			}
+			catch (ParseException ex) {
+				throw new IllegalStateException(ex);
+			}
+		}
+		return MessageBuilder.createMessage(unquoted(textPayload), headers);
+	}
+
+	private Object unquoted(Object value) {
+		String textPayload = value instanceof byte[] ? new String((byte[]) value)
+				: value.toString();
+		if (textPayload.startsWith("\"") && textPayload.endsWith("\"")) {
+			return textPayload.substring(1, textPayload.length() - 1).replace("\\\"",
+					"\"");
+		}
+		return textPayload;
+	}
+
+}

spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessagesInitializer.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2013-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.cloud.contract.verifier.messaging.kafka;
+
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.Consumer;
+
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+
+/**
+ * Logic used to initialize {@link KafkaStubMessages}. This interface might have a
+ * different implementation for the producer side and for the consumer side. That's
+ * because you can't poll for a single message by different consumers.
+ *
+ * @author Marcin Grzejszczak
+ * @since 2.2.0
+ */
+public interface KafkaStubMessagesInitializer {
+
+	/**
+	 * @param broker - embedded Kafka broker
+	 * @param kafkaProperties - kafka properties
+	 * @return topic to initialized consumer mapping
+	 */
+	Map<String, Consumer> initialize(EmbeddedKafkaBroker broker,
+			KafkaProperties kafkaProperties);
+
+}

spring-cloud-contract-verifier/src/main/resources/META-INF/spring.factories

@@ -6,4 +6,5 @@ org.springframework.cloud.contract.verifier.messaging.amqp.ContractVerifierAmqpA
 org.springframework.cloud.contract.verifier.messaging.amqp.RabbitMockConnectionFactoryAutoConfiguration,\
 org.springframework.cloud.contract.verifier.messaging.camel.ContractVerifierCamelConfiguration,\
 org.springframework.cloud.contract.verifier.messaging.jms.ContractVerifierJmsConfiguration,\
+org.springframework.cloud.contract.verifier.messaging.kafka.ContractVerifierKafkaConfiguration,\
 org.springframework.cloud.contract.verifier.messaging.noop.NoOpContractVerifierAutoConfiguration

tests/pom.xml

@@ -58,6 +58,7 @@
 				<module>spring-cloud-contract-stub-runner-stream</module>
 				<module>spring-cloud-contract-stub-runner-amqp</module>
 				<module>spring-cloud-contract-stub-runner-jms</module>
+				<module>spring-cloud-contract-stub-runner-kafka</module>
 			</modules>
 		</profile>
 		<profile>
@@ -83,6 +84,7 @@
 				<module>spring-cloud-contract-stub-runner-stream</module>
 				<module>spring-cloud-contract-stub-runner-amqp</module>
 				<module>spring-cloud-contract-stub-runner-jms</module>
+				<module>spring-cloud-contract-stub-runner-kafka</module>
 			</modules>
 		</profile>
 	</profiles>

tests/spring-cloud-contract-stub-runner-kafka/pom.xml

@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xmlns="http://maven.apache.org/POM/4.0.0"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-contract-tests</artifactId>
+		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+	<artifactId>spring-cloud-contract-stub-runner-kafka</artifactId>
+	<packaging>jar</packaging>
+	<name>Spring Cloud Contract Stub Runner JMS</name>
+	<description>Spring Cloud Contract Stub Runner JMS</description>
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-contract-stub-runner</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-contract-stub-runner-jetty</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.codehaus.groovy</groupId>
+			<artifactId>groovy</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.spockframework</groupId>
+			<artifactId>spock-core</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.spockframework</groupId>
+			<artifactId>spock-spring</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>info.solidsoft.spock</groupId>
+			<artifactId>spock-global-unroll</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.codehaus.gmavenplus</groupId>
+				<artifactId>gmavenplus-plugin</artifactId>
+			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-surefire-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+</project>

tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/BookReturned.groovy

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2013-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.cloud.contract.stubrunner.messaging.kafka
+
+
+import groovy.transform.CompileStatic
+import groovy.transform.EqualsAndHashCode
+
+@CompileStatic
+@EqualsAndHashCode
+class BookReturned implements Serializable {
+	String bookName
+
+	BookReturned(String bookName) {
+		this.bookName = bookName
+	}
+
+	BookReturned() {
+	}
+}

tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy

@@ -0,0 +1,311 @@
+/*
+ * Copyright 2013-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.cloud.contract.stubrunner.messaging.kafka
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import spock.lang.IgnoreIf
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration
+import org.springframework.boot.test.context.SpringBootContextLoader
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.cloud.contract.spec.Contract
+import org.springframework.cloud.contract.stubrunner.StubFinder
+import org.springframework.cloud.contract.stubrunner.spring.AutoConfigureStubRunner
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.context.annotation.Configuration
+import org.springframework.kafka.annotation.EnableKafka
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.support.DefaultKafkaHeaderMapper
+import org.springframework.kafka.test.EmbeddedKafkaBroker
+import org.springframework.kafka.test.context.EmbeddedKafka
+import org.springframework.messaging.Message
+import org.springframework.messaging.MessageHeaders
+import org.springframework.messaging.support.MessageBuilder
+import org.springframework.test.context.ContextConfiguration
+/**
+ * @author Marcin Grzejszczak
+ */
+@ContextConfiguration(classes = Config, loader = SpringBootContextLoader)
+@SpringBootTest(properties = ["debug=true"])
+@AutoConfigureStubRunner
+@IgnoreIf({ os.windows })
+@EmbeddedKafka(topics = ["input", "output", "delete"])
+class KafkaStubRunnerSpec extends Specification {
+
+	@Autowired
+	StubFinder stubFinder
+	@Autowired
+	KafkaTemplate kafkaTemplate
+	@Autowired
+	EmbeddedKafkaBroker broker
+	@Value('${spring.embedded.kafka.brokers}')
+	String brokers
+	@Autowired
+	MyMessageListener myMessageListener
+	PollingConditions await = new PollingConditions()
+
+	def setup() {
+		this.myMessageListener.clear()
+	}
+
+	private Message receiveFromOutput() {
+		this.myMessageListener.await()
+		return this.myMessageListener.output
+	}
+
+	def 'should download the stub and register a route for it'() {
+		when:
+			// tag::client_send[]
+			Message message = MessageBuilder.createMessage(new BookReturned('foo'), new MessageHeaders([sample: "header",]))
+			kafkaTemplate.setDefaultTopic('input')
+			kafkaTemplate.send(message)
+			// end::client_send[]
+		then:
+			// tag::client_receive[]
+			Message receivedMessage = receiveFromOutput()
+			// end::client_receive[]
+		and:
+			await.eventually {
+				// tag::client_receive_message[]
+				assert receivedMessage != null
+				assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload())
+				assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo'
+				// end::client_receive_message[]
+			}
+	}
+
+	def 'should trigger a message by label'() {
+		when:
+			// tag::client_trigger[]
+			stubFinder.trigger('return_book_1')
+			// end::client_trigger[]
+		then:
+			// tag::client_trigger_receive[]
+			Message receivedMessage = receiveFromOutput()
+			// end::client_trigger_receive[]
+		and:
+			await.eventually {
+				// tag::client_trigger_message[]
+				assert receivedMessage != null
+				assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload())
+				assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo'
+				// end::client_trigger_message[]
+			}
+	}
+
+	def 'should trigger a label for the existing groupId:artifactId'() {
+		when:
+			// tag::trigger_group_artifact[]
+			stubFinder.
+					trigger('my:stubs', 'return_book_1')
+			// end::trigger_group_artifact[]
+		then:
+			Message receivedMessage = receiveFromOutput()
+		and:
+			await.eventually {
+				assert receivedMessage != null
+				assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload())
+				assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo'
+			}
+	}
+
+	def 'should trigger a label for the existing artifactId'() {
+		when:
+			// tag::trigger_artifact[]
+			stubFinder.trigger('stubs', 'return_book_1')
+			// end::trigger_artifact[]
+		then:
+			Message receivedMessage = receiveFromOutput()
+		and:
+			await.eventually {
+				assert receivedMessage != null
+				assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload())
+				assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo'
+			}
+	}
+
+	def 'should throw an exception when missing label is passed'() {
+		when:
+			stubFinder.trigger('missing label')
+		then:
+			thrown(IllegalArgumentException)
+	}
+
+	def 'should throw an exception when missing label and artifactid is passed'() {
+		when:
+			stubFinder.trigger('some:service', 'return_book_1')
+		then:
+			thrown(IllegalArgumentException)
+	}
+
+	def 'should trigger messages by running all triggers'() {
+		when:
+			// tag::trigger_all[]
+			stubFinder.trigger()
+			// end::trigger_all[]
+		then:
+			Message receivedMessage = receiveFromOutput()
+		and:
+			await.eventually {
+				assert receivedMessage != null
+				assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload())
+				assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo'
+			}
+	}
+
+	def 'should trigger a label with no output message'() {
+		when:
+			// tag::trigger_no_output[]
+			Message message = MessageBuilder.createMessage(new BookReturned('foo'), new MessageHeaders([sample: "header",]))
+			kafkaTemplate.setDefaultTopic('delete')
+			kafkaTemplate.send(message)
+			// end::trigger_no_output[]
+		then:
+			noExceptionThrown()
+	}
+
+	def 'should not trigger a message that does not match input'() {
+		when:
+			Message message = MessageBuilder.createMessage(new BookReturned('notmatching'), new MessageHeaders([wrong: "header",]))
+			kafkaTemplate.setDefaultTopic('input')
+			kafkaTemplate.send(message)
+		then:
+			Message receivedMessage = receiveFromOutput()
+		and:
+			receivedMessage == null
+	}
+
+	private boolean assertThatBodyContainsBookNameFoo(Object payload) {
+		println "Got payload [" + payload + "]"
+		String objectAsString = payload instanceof String ? payload :
+				JsonOutput.toJson(payload)
+		def json = new JsonSlurper().parseText(objectAsString)
+		if (objectAsString.contains("payload")) {
+			return json.payload.bookName == 'foo'
+		}
+		return json.bookName == 'foo'
+	}
+
+	@Configuration
+	@ComponentScan
+	@EnableAutoConfiguration
+	@EnableKafka
+	static class Config {
+
+		@Bean
+		DefaultKafkaHeaderMapper headerMapper() {
+			return new DefaultKafkaHeaderMapper();
+		}
+
+		@Bean
+		MyMessageListener myMessageListener() {
+			return new MyMessageListener();
+		}
+	}
+
+	static class MyMessageListener {
+
+		CountDownLatch latch = new CountDownLatch(1)
+
+		Message output
+
+		@KafkaListener(topics = ["output"])
+		void output(Message message) {
+			println "I got the message [${message}]"
+			this.output = message
+			this.latch.countDown()
+		}
+
+		void clear() {
+			this.output = null
+			this.latch = new CountDownLatch(1)
+		}
+
+		void await() {
+			this.latch.await(5, TimeUnit.SECONDS)
+		}
+	}
+
+	Contract dsl =
+			// tag::sample_dsl[]
+			Contract.make {
+				label 'return_book_1'
+				input {
+					triggeredBy('bookReturnedTriggered()')
+				}
+				outputMessage {
+					sentTo('output')
+					body('''{ "bookName" : "foo" }''')
+					headers {
+						header('BOOK-NAME', 'foo')
+					}
+				}
+			}
+	// end::sample_dsl[]
+
+	Contract dsl2 =
+			// tag::sample_dsl_2[]
+			Contract.make {
+				label 'return_book_2'
+				input {
+					messageFrom('input')
+					messageBody([
+							bookName: 'foo'
+					])
+					messageHeaders {
+						header('sample', 'header')
+					}
+				}
+				outputMessage {
+					sentTo('output')
+					body([
+							bookName: 'foo'
+					])
+					headers {
+						header('BOOK-NAME', 'foo')
+					}
+				}
+			}
+	// end::sample_dsl_2[]
+
+	Contract dsl3 =
+			// tag::sample_dsl_3[]
+			Contract.make {
+				label 'delete_book'
+				input {
+					messageFrom('delete')
+					messageBody([
+							bookName: 'foo'
+					])
+					messageHeaders {
+						header('sample', 'header')
+					}
+					assertThat('bookWasDeleted()')
+				}
+			}
+	// end::sample_dsl_3[]
+}

tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/application.yml

@@ -0,0 +1,21 @@
+stubrunner:
+  repository-root: stubs:classpath:/stubs/
+  ids: my:stubs
+  stubs-mode: remote
+spring:
+  kafka:
+    bootstrap-servers: ${spring.embedded.kafka.brokers}
+    producer:
+      properties:
+        "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer"
+        "spring.json.trusted.packages": "*"
+    consumer:
+      properties:
+        "value.deserializer": "org.springframework.kafka.support.serializer.JsonDeserializer"
+        "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer"
+        "spring.json.trusted.packages": "*"
+      group-id: groupId
+server:
+  port: 0
+debug: true
+logging.level.org.springframework.cloud.contract: debug

tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookDeleted.groovy

@@ -0,0 +1,13 @@
+org.springframework.cloud.contract.spec.Contract.make {
+	label 'delete_book'
+	input {
+		messageFrom('delete')
+		messageBody([
+				bookName: 'foo'
+		])
+		messageHeaders {
+			header('sample', 'header')
+		}
+		assertThat('bookWasDeleted()')
+	}
+}

tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned1.groovy

@@ -0,0 +1,13 @@
+org.springframework.cloud.contract.spec.Contract.make {
+	label 'return_book_1'
+	input {
+		triggeredBy('bookReturnedTriggered()')
+	}
+	outputMessage {
+		sentTo('output')
+		body('''{ "bookName" : "foo" }''')
+		headers {
+			header('BOOK-NAME', 'foo')
+		}
+	}
+}

tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned2.groovy

@@ -0,0 +1,21 @@
+org.springframework.cloud.contract.spec.Contract.make {
+	label 'return_book_2'
+	input {
+		messageFrom('input')
+		messageBody([
+				bookName: 'foo'
+		])
+		messageHeaders {
+			header('sample', 'header')
+		}
+	}
+	outputMessage {
+		sentTo('output')
+		body([
+				bookName: 'foo'
+		])
+		headers {
+			header('BOOK-NAME', 'foo')
+		}
+	}
+}