Fix Kafka Middleware sample config.

Author: OlgaMaciaszek

consumer_kafka_middleware/src/test/java/com/example/ApplicationTests.java

@@ -19,7 +19,6 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 import javax.annotation.Nullable;
 
@@ -36,12 +35,13 @@
 import org.springframework.cloud.contract.stubrunner.spring.AutoConfigureStubRunner;
 import org.springframework.cloud.contract.stubrunner.spring.StubRunnerProperties;
 import org.springframework.cloud.contract.verifier.converter.YamlContract;
-import org.springframework.cloud.contract.verifier.messaging.MessageVerifierReceiver;
 import org.springframework.cloud.contract.verifier.messaging.MessageVerifierSender;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.converter.JsonMessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageBuilder;
@@ -55,7 +55,7 @@
 @ActiveProfiles("test")
 public class ApplicationTests {
 
-	
+
 	@Container
 	static KafkaContainer kafka = new KafkaContainer();
 
@@ -79,7 +79,7 @@ public void contextLoads() {
 			BDDAssertions.then(this.application.storedFoo.getFoo()).contains("example");
 		});
 	}
-	
+
 }
 
 
@@ -102,4 +102,21 @@ public <T> void send(T payload, Map<String, Object> headers, String destination,
 			}
 		};
 	}
+
+	@Bean
+	@Primary
+	JsonMessageConverter noopMessageConverter() {
+		return new NoopJsonMessageConverter();
+	}
+}
+
+class NoopJsonMessageConverter extends JsonMessageConverter {
+
+	NoopJsonMessageConverter() {
+	}
+
+	@Override
+	protected Object convertPayload(Message<?> message) {
+		return message.getPayload();
+	}
 }

consumer_kafka_middleware/src/test/resources/application-test.yml

@@ -1,5 +1,7 @@
 spring:
   kafka:
+    producer:
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
     consumer:
       properties:
         "key.serializer": "org.springframework.kafka.support.serializer.JsonSerializer"

producer_kafka_middleware/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>3.0.0-SNAPSHOT</version>
+		<version>3.0.6-SNAPSHOT</version>
 		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 
@@ -42,13 +42,11 @@
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>junit-jupiter</artifactId>
-			<version>1.17.1</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>kafka</artifactId>
-			<version>1.17.1</version>
 			<scope>test</scope>
 		</dependency>
 		<!-- spring cloud contract test dependencies -->
@@ -65,8 +63,9 @@
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 		<java.version>17</java.version>
-		<spring-cloud.version>2022.0.0-SNAPSHOT</spring-cloud.version>
-		<spring-cloud-contract.version>4.0.0-SNAPSHOT</spring-cloud-contract.version>
+		<spring-cloud.version>2022.0.3-SNAPSHOT</spring-cloud.version>
+		<spring-cloud-contract.version>4.0.3-SNAPSHOT</spring-cloud-contract.version>
+		<testcontainers.version>1.17.6</testcontainers.version>
 	</properties>
 
 	<build>
@@ -107,6 +106,13 @@
 				<type>pom</type>
 				<scope>import</scope>
 			</dependency>
+			<dependency>
+				<groupId>org.testcontainers</groupId>
+				<artifactId>testcontainers-bom</artifactId>
+				<version>${testcontainers.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 

producer_kafka_middleware/src/test/java/com/example/BaseClass.java

@@ -26,7 +26,6 @@
 
 import javax.annotation.Nullable;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,9 +38,6 @@
 import org.springframework.cloud.contract.verifier.converter.YamlContract;
 import org.springframework.cloud.contract.verifier.messaging.MessageVerifierReceiver;
 import org.springframework.cloud.contract.verifier.messaging.boot.AutoConfigureMessageVerifier;
-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.internal.ContractVerifierObjectMapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
@@ -51,20 +47,20 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.handler.annotation.Header;
-import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.DynamicPropertyRegistry;
 import org.springframework.test.context.DynamicPropertySource;
 
-@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, classes = { TestConfig.class, Application.class })
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, classes = {TestConfig.class, Application.class})
 @Testcontainers
 @AutoConfigureMessageVerifier
 @ActiveProfiles("test")
 public abstract class BaseClass {
-	
 
-	@Container static KafkaContainer kafka = new KafkaContainer();
+
+	@Container
+	static KafkaContainer kafka = new KafkaContainer();
 
 	@DynamicPropertySource
 	static void kafkaProperties(DynamicPropertyRegistry registry) {
@@ -75,9 +71,9 @@ static void kafkaProperties(DynamicPropertyRegistry registry) {
 	Controller controller;
 
 	public void trigger() {
-		
+
 		this.controller.sendFoo("example");
-		
+
 	}
 }
 
@@ -114,7 +110,8 @@ public Message receive(String destination, long timeout, TimeUnit timeUnit, @Nul
 	private void await(long timeout, TimeUnit timeUnit) {
 		try {
 			cyclicBarrier.await(timeout, timeUnit);
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 
 		}
 	}