Avro support

pom.xml

@@ -626,6 +626,14 @@
 				<enabled>true</enabled>
 			</snapshots>
 		</repository>
+		<repository>
+			<id>confluent</id>
+			<name>Confluent</name>
+			<url>https://packages.confluent.io/maven/</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</repository>
 		<repository>
 			<id>netflix-candidates</id>
 			<name>Netflix Candidates</name>

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

@@ -174,6 +174,12 @@
 			<version>4.0.2</version>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.avro</groupId>
+			<artifactId>avro</artifactId>
+			<version>1.12.1</version>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>cglib</groupId>
 			<artifactId>cglib</artifactId>

spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/StubRunnerExecutor.java

@@ -251,7 +251,7 @@ private void sendMessage(Contract groovyDsl) {
 		setMessageType(contract, ContractVerifierMessageMetadata.MessageType.OUTPUT);
 
 		Object payload = null;
-		if (body != null && body.getClientValue() instanceof FromFileProperty) {
+		if (isFromFileProperty(body)) {
 			FromFileProperty fromFile = (FromFileProperty) body.getClientValue();
 			if (fromFile.isByte()) {
 				payload = fromFile.asBytes();
@@ -260,15 +260,40 @@ private void sendMessage(Contract groovyDsl) {
 				payload = fromFile.asString();
 			}
 		}
+		else if (isAvroContract(contract)) {
+			log.info("Avro contract detected — passing raw body as Map, skipping JSON serialization");
+			payload = BodyExtractor.extractClientValueFromBody(body == null ? null : body.getClientValue());
+		}
 		else {
 			payload = JsonOutput
 				.toJson(BodyExtractor.extractClientValueFromBody(body == null ? null : body.getClientValue()));
 		}
 
-		this.messageVerifierSender.send(payload, headers == null ? null : headers.asStubSideMap(),
+		this.messageVerifierSender.send(payload,
+				headers == null ? null : headers.asStubSideMap(),
 				outputMessage.getSentTo().getClientValue(), contract);
 	}
 
+	private boolean isFromFileProperty(final DslProperty<?> body) {
+		return body != null
+				&& body.getClientValue() instanceof FromFileProperty;
+	}
+
+	private boolean isAvroContract(final YamlContract contract) {
+		if (contract == null || contract.metadata == null) {
+			return false;
+		}
+		Object kafkaMeta = contract.metadata.get("kafka");
+		if (!(kafkaMeta instanceof Map)) {
+			return false;
+		}
+		Object avroMeta = ((Map<String, Object>) kafkaMeta).get("avro");
+		if (!(avroMeta instanceof Map)) {
+			return false;
+		}
+		return ((Map<String, Object>) avroMeta).get("schema") != null;
+	}
+
 	private void setMessageType(YamlContract contract, ContractVerifierMessageMetadata.MessageType output) {
 		contract.metadata.put(ContractVerifierMessageMetadata.METADATA_KEY,
 				new ContractVerifierMessageMetadata(output));

spring-cloud-contract-stub-runner/src/test/groovy/org/springframework/cloud/contract/stubrunner/StubRunnerExecutorAvroSpec.groovy

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2013-present 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
+
+import org.apache.avro.generic.GenericRecord
+import spock.lang.Specification
+
+import org.springframework.cloud.contract.verifier.messaging.avro.KafkaAvroMessageVerifierSender
+import org.springframework.kafka.core.KafkaTemplate
+
+class StubRunnerExecutorAvroSpec extends Specification {
+
+	private KafkaTemplate<String, Object> kafkaTemplate = Mock()
+	private KafkaAvroMessageVerifierSender sender = new KafkaAvroMessageVerifierSender(kafkaTemplate)
+
+	def 'should send Avro-serialized GenericRecord to Kafka for Avro contracts (bug #2404)'() {
+		given:
+			def tmpContractDir = saveTmpContract("""
+label: book_returned
+input:
+  triggeredBy: publishBookReturned()
+outputMessage:
+  sentTo: book.returned
+  headers:
+    X-Correlation-Id: abc-123-def
+  body:
+    isbn: "978-1234567890"
+    title: "Contract Testing for Dummies"
+metadata:
+  kafka:
+    avro:
+      schema: >
+        {
+          "type": "record",
+          "name": "Book",
+          "fields": [
+            {"name": "isbn", "type": "string"},
+            {"name": "title", "type": "string"}
+          ]
+        }
+""")
+			StubRunnerExecutor executor = new StubRunnerExecutor(new AvailablePortScanner(18000, 18999), sender, [])
+			executor.runStubs(
+					new StubRunnerOptionsBuilder().build(),
+					new StubRepository(tmpContractDir, [], new StubRunnerOptionsBuilder().build(), null),
+					new StubConfiguration('avro', 'avro', 'avro', ''))
+		when:
+			executor.trigger('book_returned')
+		then:
+			1 * kafkaTemplate.send({
+				it.topic() == "book.returned" &&
+				it.value() instanceof GenericRecord &&
+				it.value()["schema"] != null &&
+				it.value()["isbn"] == "978-1234567890" &&
+				it.value()["title"] == "Contract Testing for Dummies" &&
+				header(it, "X-Correlation-Id") == "abc-123-def"
+			})
+		cleanup:
+			executor.shutdown()
+			tmpContractDir.deleteDir()
+	}
+
+	private File saveTmpContract(String contractYaml) {
+		File contractDir = File.createTempDir()
+		new File(contractDir, "book_returned.yml").text = contractYaml
+		return contractDir
+	}
+
+	private String header(it, String key) {
+		new String(it.headers().lastHeader(key).value())
+	}
+
+}

spring-cloud-contract-stub-runner/src/test/resources/avro-repo/book_returned.yml

@@ -0,0 +1,22 @@
+label: book_returned
+input:
+  triggeredBy: publishBookReturned()
+outputMessage:
+  sentTo: book.returned
+  headers:
+    X-Correlation-Id: abc-123-def
+  body:
+    isbn: "978-1234567890"
+    title: "Contract Testing for Dummies"
+metadata:
+  kafka:
+    avro:
+      schema: >
+        {
+          "type": "record",
+          "name": "Book",
+          "fields": [
+            {"name": "isbn", "type": "string"},
+            {"name": "title", "type": "string"}
+          ]
+        }

spring-cloud-contract-verifier/pom.xml

@@ -5,6 +5,8 @@
 	<modelVersion>4.0.0</modelVersion>
 	<properties>
 		<javax.ws.rs-api.version>2.1.1</javax.ws.rs-api.version>
+		<avro.version>1.12.1</avro.version>
+		<kafka-avro-serializer.version>7.9.0</kafka-avro-serializer.version>
 	</properties>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
@@ -53,6 +55,18 @@
 			<artifactId>spring-kafka</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.avro</groupId>
+			<artifactId>avro</artifactId>
+			<version>${avro.version}</version>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>io.confluent</groupId>
+			<artifactId>kafka-avro-serializer</artifactId>
+			<version>${kafka-avro-serializer.version}</version>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.kafka</groupId>
 			<artifactId>spring-kafka-test</artifactId>
@@ -278,6 +292,11 @@
 			<artifactId>spock-junit4</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-resttestclient</artifactId>

spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/avro/AvroMetadata.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2013-present 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.avro;
+
+/**
+ * Avro serialization metadata for a Kafka contract message.
+ *
+ * <p>
+ * Example contract YAML: <pre>
+ * metadata:
+ *   kafka:
+ *     avro:
+ *       schema: classpath:avro/Book.avsc
+ * </pre>
+ *
+ * <p>
+ * The Schema Registry URL is configured globally via
+ * {@code spring.kafka.properties.schema.registry.url}.
+ *
+ * @author Emanuel Trandafir
+ * @since 4.2.0
+ */
+public final class AvroMetadata {
+
+    /**
+     * Classpath or filesystem path to the Avro schema file
+     * ({@code .avsc}), e.g. {@code classpath:avro/Book.avsc}.
+     * May also be an inline JSON schema string.
+     */
+    private String schema;
+
+    /**
+     * Returns the Avro schema path or inline schema JSON.
+     *
+     * @return the schema
+     */
+    public String getSchema() {
+        return this.schema;
+    }
+
+    /**
+     * Sets the Avro schema path or inline schema JSON.
+     *
+     * @param value the schema
+     */
+    public void setSchema(final String value) {
+        this.schema = value;
+    }
+
+}