SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Kafka binder tests for Boot 3.3.x compatibility

Browse Source
This commit is contained in:
Soby Chacko
2024-04-12 16:42:17 -04:00
parent 24556fd79e
commit 5ab62e1aa1
1 changed files with 35 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -52,7 +52,6 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
@@ -66,8 +65,6 @@ import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.assertj.core.api.Assertions;
import org.assertj.core.api.Condition;
@@ -159,7 +156,6 @@ import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.backoff.FixedBackOff;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.entry;
@@ -3172,7 +3168,7 @@ class KafkaBinderTests extends
}
@Test
@SuppressWarnings("unchecked")
@SuppressWarnings({"unchecked", "rawtypes"})
void consumerDefaultDeserializer() throws Throwable {
Binding<?> binding = null;
try {
@@ -3188,13 +3184,21 @@ class KafkaBinderTests extends
binding = binder.bindConsumer(testTopicName, "test", input,
consumerProperties);
var consumerAccessor = new DirectFieldAccessor(
getKafkaConsumer(binding));
assertThat(consumerAccessor
.getPropertyValue("keyDeserializer") instanceof ByteArrayDeserializer)
.isTrue();
assertThat(consumerAccessor.getPropertyValue(
"valueDeserializer") instanceof ByteArrayDeserializer).isTrue();
var bindingAccessor = new DirectFieldAccessor(binding);
KafkaMessageDrivenChannelAdapter adapter = (KafkaMessageDrivenChannelAdapter) bindingAccessor
.getPropertyValue("lifecycle");
var adapterAccessor = new DirectFieldAccessor(adapter);
ConcurrentMessageListenerContainer messageListenerContainer = (ConcurrentMessageListenerContainer) adapterAccessor
.getPropertyValue("messageListenerContainer");
var containerAccessor = new DirectFieldAccessor(
messageListenerContainer);
DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) containerAccessor
.getPropertyValue("consumerFactory");
Map<String, Object> configProps = consumerFactory.getConfigurationProperties();
assertThat(configProps.get("key.deserializer")).isEqualTo(org.apache.kafka.common.serialization.ByteArrayDeserializer.class);
assertThat(configProps.get("value.deserializer")).isEqualTo(org.apache.kafka.common.serialization.ByteArrayDeserializer.class);
}
finally {
if (binding != null) {
@@ -3204,7 +3208,7 @@ class KafkaBinderTests extends
}
@Test
@SuppressWarnings("unchecked")
@SuppressWarnings({"unchecked", "rawtypes"})
void consumerCustomDeserializer() throws Exception {
Binding<?> binding = null;
try {
@@ -3226,14 +3230,23 @@ class KafkaBinderTests extends
binding = binder.bindConsumer(testTopicName, "test", input,
consumerProperties);
var consumerAccessor = new DirectFieldAccessor(
getKafkaConsumer(binding));
assertThat(consumerAccessor
.getPropertyValue("keyDeserializer") instanceof StringDeserializer)
.isTrue();
assertThat(consumerAccessor
.getPropertyValue("valueDeserializer") instanceof LongDeserializer)
.isTrue();
var bindingAccessor = new DirectFieldAccessor(binding);
KafkaMessageDrivenChannelAdapter adapter = (KafkaMessageDrivenChannelAdapter) bindingAccessor
.getPropertyValue("lifecycle");
var adapterAccessor = new DirectFieldAccessor(adapter);
ConcurrentMessageListenerContainer messageListenerContainer = (ConcurrentMessageListenerContainer) adapterAccessor
.getPropertyValue("messageListenerContainer");
var containerAccessor = new DirectFieldAccessor(
messageListenerContainer);
DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) containerAccessor
.getPropertyValue("consumerFactory");
Map<String, Object> configProps = consumerFactory.getConfigurationProperties();
assertThat(configProps.get("key.deserializer")).isEqualTo("org.apache.kafka.common.serialization.StringDeserializer");
assertThat(configProps.get("value.deserializer")).isEqualTo("org.apache.kafka.common.serialization.LongDeserializer");
}
finally {
if (binding != null) {
@@ -3291,20 +3304,6 @@ class KafkaBinderTests extends
}
}
private KafkaConsumer getKafkaConsumer(Binding binding) {
var bindingAccessor = new DirectFieldAccessor(binding);
KafkaMessageDrivenChannelAdapter adapter = (KafkaMessageDrivenChannelAdapter) bindingAccessor
.getPropertyValue("lifecycle");
var adapterAccessor = new DirectFieldAccessor(adapter);
ConcurrentMessageListenerContainer messageListenerContainer = (ConcurrentMessageListenerContainer) adapterAccessor
.getPropertyValue("messageListenerContainer");
var containerAccessor = new DirectFieldAccessor(
messageListenerContainer);
DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) containerAccessor
.getPropertyValue("consumerFactory");
return (KafkaConsumer) consumerFactory.createConsumer();
}
@Test
@SuppressWarnings({ "unchecked", "rawtypes" })
void nativeSerializationWithCustomSerializerDeserializerBytesPayload()