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

GH-2685: Reactive Kafka Customizers - Add Generics

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2685

Required to allow customization of serializers/deserializers.
This commit is contained in:
Gary Russell
2023-04-03 12:39:08 -04:00
committed by Soby Chacko
parent b814c0f395
commit 64835ef1d7
5 changed files with 27 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java
View File

@@ -92,9 +92,9 @@ public class ReactorKafkaBinder
private ProducerConfigCustomizer producerConfigCustomizer;
private ReceiverOptionsCustomizer receiverOptionsCustomizer = (name, opts) -> opts;
private ReceiverOptionsCustomizer<Object, Object> receiverOptionsCustomizer = (name, opts) -> opts;
private SenderOptionsCustomizer senderOptionsCustomizer = (name, opts) -> opts;
private SenderOptionsCustomizer<Object, Object> senderOptionsCustomizer = (name, opts) -> opts;
public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties,
KafkaTopicProvisioner provisioner) {
@@ -111,6 +111,7 @@ public class ReactorKafkaBinder
this.producerConfigCustomizer = producerConfigCustomizer;
}
@SuppressWarnings({ "rawtypes", "unchecked" })
public void receiverOptionsCustomizers(ObjectProvider<ReceiverOptionsCustomizer> customizers) {
if (customizers.getIfUnique() != null) {
this.receiverOptionsCustomizer = customizers.getIfUnique();
@@ -120,7 +121,7 @@ public class ReactorKafkaBinder
ReceiverOptionsCustomizer customizer = (name, opts) -> {
ReceiverOptions<Object, Object> last = null;
for (ReceiverOptionsCustomizer cust: list) {
last = cust.apply(name, opts);
last = (ReceiverOptions<Object, Object>) cust.apply(name, opts);
}
return last;
};
@@ -130,6 +131,7 @@ public class ReactorKafkaBinder
}
}
@SuppressWarnings({ "rawtypes", "unchecked" })
public void senderOptionsCustomizers(ObjectProvider<SenderOptionsCustomizer> customizers) {
if (customizers.getIfUnique() != null) {
this.senderOptionsCustomizer = customizers.getIfUnique();
@@ -139,7 +141,7 @@ public class ReactorKafkaBinder
SenderOptionsCustomizer customizer = (name, opts) -> {
SenderOptions<Object, Object> last = null;
for (SenderOptionsCustomizer cust: list) {
last = cust.apply(name, opts);
last = (SenderOptions<Object, Object>) cust.apply(name, opts);
}
return last;
};

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java
View File

@@ -27,13 +27,18 @@ import org.springframework.core.Ordered;
* the second is the {@link ReceiverOptions} to customize. Return the customized
* {@link ReceiverOptions}. Applied in {@link Ordered order} if multiple customizers are
* found.
* <p>
* Generic since 4.0.3 to allow customization of deserializers.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Gary Russell
* @since 4.0.2
*
*/
public interface ReceiverOptionsCustomizer
extends BiFunction<String, ReceiverOptions<Object, Object>, ReceiverOptions<Object, Object>>, Ordered {
public interface ReceiverOptionsCustomizer<K, V>
extends BiFunction<String, ReceiverOptions<K, V>, ReceiverOptions<K, V>>, Ordered {
@Override
default int getOrder() {

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java
View File

@@ -27,13 +27,18 @@ import org.springframework.core.Ordered;
* second is the {@link SenderOptions} to customize. Return the customized
* {@link SenderOptions}. Applied in {@link Ordered order} if multiple customizers are
* found.
* <p>
* Generic since 4.0.3 to allow customization of serializers.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Gary Russell
* @since 4.0.2
*
*/
public interface SenderOptionsCustomizer
extends BiFunction<String, SenderOptions<Object, Object>, SenderOptions<Object, Object>>, Ordered {
public interface SenderOptionsCustomizer<K, V>
extends BiFunction<String, SenderOptions<K, V>, SenderOptions<K, V>>, Ordered {
@Override
default int getOrder() {

11
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java
View File

@@ -26,6 +26,7 @@ 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.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -167,7 +168,7 @@ class ReactorKafkaBinderIntegrationTests {
}
@Bean
ReceiverOptionsCustomizer cust1() {
ReceiverOptionsCustomizer<?, ?> cust1() {
return (t, u) -> {
recOptsCustOrder.add("one");
return u;
@@ -175,12 +176,13 @@ class ReactorKafkaBinderIntegrationTests {
}
@Bean
ReceiverOptionsCustomizer cust2() {
return new ReceiverOptionsCustomizer() {
ReceiverOptionsCustomizer<String, byte[]> cust2() {
return new ReceiverOptionsCustomizer<>() {
@Override
public ReceiverOptions<Object, Object> apply(String t, ReceiverOptions<Object, Object> u) {
public ReceiverOptions<String, byte[]> apply(String t, ReceiverOptions<String, byte[]> u) {
recOptsCustOrder.add("two");
u.withKeyDeserializer(new StringDeserializer());
return u;
}
@@ -189,7 +191,6 @@ class ReactorKafkaBinderIntegrationTests {
return -1;
}
};
}

1
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java
View File

@@ -220,6 +220,7 @@ public class ReactorKafkaBinderTests {
provisioner.setMetadataRetryOperations(new RetryTemplate());
ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner);
binder.setApplicationContext(new GenericApplicationContext());
@SuppressWarnings("rawtypes")
ObjectProvider<SenderOptionsCustomizer> cust = mock(ObjectProvider.class);
AtomicBoolean custCalled = new AtomicBoolean();
given(cust.getIfUnique()).willReturn((name, opts) -> {