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

Kafka Streams binder code cleanup

Browse Source 
- Remove deprecated API usage
 - Other cleanup
 - EmbeddedKafka changes in tests
This commit is contained in:
Soby Chacko
2023-10-09 21:12:27 -04:00
parent 93c4fe39ae
commit d650f9e1e8
24 changed files with 135 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java
View File

@@ -94,7 +94,7 @@ public class GlobalKTableBinder extends
this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget),
this.kafkaStreamsBindingInformationCatalogue, streamsBuilderFactoryBean);
return new DefaultBinding<GlobalKTable<Object, Object>>(bindingName, group, inputTarget, streamsBuilderFactoryBean) {
return new DefaultBinding<>(bindingName, group, inputTarget, streamsBuilderFactoryBean) {
@Override
public boolean isInput() {
@@ -125,7 +125,7 @@ public class GlobalKTableBinder extends
//Caching the stopped KafkaStreams for health indicator purposes on the underlying processor.
//See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1165
GlobalKTableBinder.this.kafkaStreamsBindingInformationCatalogue.addPreviousKafkaStreamsForApplicationId(
(String) streamsBuilderFactoryBean.getStreamsConfiguration().get(StreamsConfig.APPLICATION_ID_CONFIG), kafkaStreams);
(String) streamsBuilderFactoryBean.getStreamsConfiguration().get(StreamsConfig.APPLICATION_ID_CONFIG), kafkaStreams);
}
}
};

12
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java
View File

@@ -31,11 +31,11 @@ import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyQueryMetadata;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetadata;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.errors.UnknownStateStoreException;
import org.apache.kafka.streams.state.HostInfo;
import org.apache.kafka.streams.state.QueryableStoreType;
import org.apache.kafka.streams.state.StreamsMetadata;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
import org.springframework.retry.RetryPolicy;
@@ -171,8 +171,8 @@ public class InteractiveQueryService {
* @return true if Streams Instance is associated with Thread
*/
private boolean filterByThreadName(KafkaStreams streams) {
String applicationId = kafkaStreamsRegistry.streamBuilderFactoryBean(
streams).getStreamsConfiguration()
String applicationId = Objects.requireNonNull(kafkaStreamsRegistry.streamBuilderFactoryBean(
streams).getStreamsConfiguration())
.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
// TODO: is there some better way to find out if a Stream App created the Thread?
return Thread.currentThread().getName().contains(applicationId);
@@ -195,7 +195,7 @@ public class InteractiveQueryService {
String applicationServer = configuration.get("application.server");
String[] splits = StringUtils.split(applicationServer, ":");
return new HostInfo(splits[0], Integer.parseInt(splits[1]));
return new HostInfo(Objects.requireNonNull(splits)[0], Integer.parseInt(splits[1]));
}
return null;
}
@@ -306,7 +306,7 @@ public class InteractiveQueryService {
public List<HostInfo> getAllHostsInfo(String store) {
return kafkaStreamsRegistry.getKafkaStreams()
.stream()
.flatMap(k -> k.allMetadataForStore(store).stream())
.flatMap(k -> k.streamsMetadataForStore(store).stream())
.filter(Objects::nonNull)
.map(StreamsMetadata::hostInfo)
.collect(Collectors.toList());
@@ -316,7 +316,7 @@ public class InteractiveQueryService {
* @param storeQueryParametersCustomizer to customize
* @since 4.0.1
*/
public void setStoreQueryParametersCustomizer(StoreQueryParametersCustomizer storeQueryParametersCustomizer) {
public void setStoreQueryParametersCustomizer(StoreQueryParametersCustomizer<?> storeQueryParametersCustomizer) {
this.storeQueryParametersCustomizer = storeQueryParametersCustomizer;
}
}

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.Objects;
import java.util.Properties;
import org.apache.commons.logging.Log;
@@ -137,7 +138,7 @@ class KStreamBinder extends
KStreamBinder.this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean);
//If we cached the previous KafkaStreams object (from a binding stop on the actuator), remove it.
//See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1165
final String applicationId = (String) streamsBuilderFactoryBean.getStreamsConfiguration().get(StreamsConfig.APPLICATION_ID_CONFIG);
final String applicationId = (String) Objects.requireNonNull(streamsBuilderFactoryBean.getStreamsConfiguration()).get(StreamsConfig.APPLICATION_ID_CONFIG);
if (kafkaStreamsBindingInformationCatalogue.getStoppedKafkaStreams().containsKey(applicationId)) {
kafkaStreamsBindingInformationCatalogue.removePreviousKafkaStreamsForApplicationId(applicationId);
}
@@ -154,7 +155,7 @@ class KStreamBinder extends
//Caching the stopped KafkaStreams for health indicator purposes on the underlying processor.
//See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1165
KStreamBinder.this.kafkaStreamsBindingInformationCatalogue.addPreviousKafkaStreamsForApplicationId(
(String) streamsBuilderFactoryBean.getStreamsConfiguration().get(StreamsConfig.APPLICATION_ID_CONFIG), kafkaStreams);
(String) Objects.requireNonNull(streamsBuilderFactoryBean.getStreamsConfiguration()).get(StreamsConfig.APPLICATION_ID_CONFIG), kafkaStreams);
}
}
};

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java
View File

@@ -16,6 +16,8 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.Objects;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KTable;
@@ -36,6 +38,7 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.StringUtils;
/**
* {@link org.springframework.cloud.stream.binder.Binder} implementation for
* {@link KTable}. This implemenation extends from the {@link AbstractBinder} directly.
@@ -73,7 +76,6 @@ class KTableBinder extends
}
@Override
@SuppressWarnings("unchecked")
protected Binding<KTable<Object, Object>> doBindConsumer(String name, String group,
KTable<Object, Object> inputTarget,
// @checkstyle:off
@@ -95,7 +97,7 @@ class KTableBinder extends
this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget),
this.kafkaStreamsBindingInformationCatalogue, streamsBuilderFactoryBean);
return new DefaultBinding<KTable<Object, Object>>(bindingName, group, inputTarget, streamsBuilderFactoryBean) {
return new DefaultBinding<>(bindingName, group, inputTarget, streamsBuilderFactoryBean) {
@Override
public boolean isInput() {
@@ -109,7 +111,7 @@ class KTableBinder extends
KTableBinder.this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean);
//If we cached the previous KafkaStreams object (from a binding stop on the actuator), remove it.
//See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1165
final String applicationId = (String) streamsBuilderFactoryBean.getStreamsConfiguration().get(StreamsConfig.APPLICATION_ID_CONFIG);
final String applicationId = (String) Objects.requireNonNull(streamsBuilderFactoryBean.getStreamsConfiguration()).get(StreamsConfig.APPLICATION_ID_CONFIG);
if (kafkaStreamsBindingInformationCatalogue.getStoppedKafkaStreams().containsKey(applicationId)) {
kafkaStreamsBindingInformationCatalogue.removePreviousKafkaStreamsForApplicationId(applicationId);
}
@@ -126,7 +128,7 @@ class KTableBinder extends
//Caching the stopped KafkaStreams for health indicator purposes on the underlying processor.
//See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1165
KTableBinder.this.kafkaStreamsBindingInformationCatalogue.addPreviousKafkaStreamsForApplicationId(
(String) streamsBuilderFactoryBean.getStreamsConfiguration().get(StreamsConfig.APPLICATION_ID_CONFIG), kafkaStreams);
(String) Objects.requireNonNull(streamsBuilderFactoryBean.getStreamsConfiguration()).get(StreamsConfig.APPLICATION_ID_CONFIG), kafkaStreams);
}
}
};

2
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java
View File

@@ -55,7 +55,7 @@ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator i
/**
* Static initialization for detecting whether the application is using Kafka client 2.5 vs lower versions.
*/
private static ClassLoader CLASS_LOADER = KafkaStreamsBinderHealthIndicator.class.getClassLoader();
private static final ClassLoader CLASS_LOADER = KafkaStreamsBinderHealthIndicator.class.getClassLoader();
private static boolean isKafkaStreams25 = true;
private static Method methodForIsRunning;

3
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java
View File

@@ -20,6 +20,7 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -94,7 +95,7 @@ public class KafkaStreamsBinderMetrics {
for (StreamsBuilderFactoryBean streamsBuilderFactoryBean : streamsBuilderFactoryBeans) {
if (streamsBuilderFactoryBean.isRunning()) {
KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
final Map<MetricName, ? extends Metric> metrics = kafkaStreams.metrics();
final Map<MetricName, ? extends Metric> metrics = Objects.requireNonNull(kafkaStreams).metrics();
prepareToBindMetrics(registry, metrics);
checkAndBindMetrics(registry, metrics);

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -153,10 +153,9 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
}
// TODO: Lifted from core - good candidate for exposing as a utility method in core.
@SuppressWarnings("unchecked")
private void flatten(String propertyName, Object value,
Map<String, Object> flattenedProperties) {
if (value instanceof Map valueAsMap) {
if (value instanceof Map<?, ?> valueAsMap) {
valueAsMap.forEach((k, v) -> flatten(
(propertyName != null ? propertyName + "." : "") + k, v,
flattenedProperties));
@@ -220,7 +219,7 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
kafkaConnectionString);
}
}
else if (bootstrapServerConfig instanceof List bootStrapCollection) {
else if (bootstrapServerConfig instanceof List<?> bootStrapCollection) {
if (bootStrapCollection.size() == 1 && bootStrapCollection.get(0).equals("localhost:9092")) {
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaConnectionString);

1
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java
View File

@@ -110,6 +110,7 @@ public final class KafkaStreamsBinderUtils {
return Arrays.stream(rawSplitDefinition).map(String::trim).toArray(String[]::new);
}
@SuppressWarnings({"unchecked", "rawtypes"})
static void prepareConsumerBinding(String name, String group,
ApplicationContext context, KafkaTopicProvisioner kafkaTopicProvisioner,
KafkaStreamsBinderConfigurationProperties binderConfigurationProperties,

17
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java
View File

@@ -24,6 +24,7 @@ import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.function.BiConsumer;
@@ -60,6 +61,7 @@ import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
import org.springframework.kafka.core.CleanupConfig;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -81,8 +83,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
private final KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate;
private BeanFactory beanFactory;
private StreamFunctionProperties streamFunctionProperties;
private KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties;
private final StreamFunctionProperties streamFunctionProperties;
private final KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties;
StreamsBuilderFactoryBeanConfigurer customizer;
ConfigurableEnvironment environment;
@@ -189,7 +191,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
}
private boolean functionOrConsumerFound(ResolvableType iterableResType) {
return iterableResType.getRawClass().equals(Function.class) ||
return Objects.requireNonNull(iterableResType.getRawClass()).equals(Function.class) ||
iterableResType.getRawClass().equals(Consumer.class);
}
@@ -505,7 +507,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
StreamsBuilderFactoryBean streamsBuilderFactoryBean =
this.methodStreamsBuilderFactoryBeanMap.get(functionName);
StreamsBuilder streamsBuilder = streamsBuilderFactoryBean.getObject();
final String applicationId = streamsBuilderFactoryBean.getStreamsConfiguration().getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
final String applicationId = Objects.requireNonNull(streamsBuilderFactoryBean.getStreamsConfiguration())
.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
KafkaStreamsConsumerProperties extendedConsumerProperties =
this.kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(input);
extendedConsumerProperties.setApplicationId(applicationId);
@@ -518,7 +521,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
LOG.info("Value Serde used for " + input + ": " + valueSerde.getClass().getName());
final Topology.AutoOffsetReset autoOffsetReset = getAutoOffsetReset(input, extendedConsumerProperties);
if (parameterType.isAssignableFrom(KStream.class)) {
if (Objects.requireNonNull(parameterType).isAssignableFrom(KStream.class)) {
KStream<?, ?> stream = getKStream(input, bindingProperties, extendedConsumerProperties,
streamsBuilder, keySerde, valueSerde, autoOffsetReset, i == 0);
KStreamBoundElementFactory.KStreamWrapper kStreamWrapper =
@@ -532,7 +535,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
this.kafkaStreamsBindingInformationCatalogue.addConsumerPropertiesPerSbfb(streamsBuilderFactoryBean,
bindingServiceProperties.getConsumerProperties(input));
if (KStream.class.isAssignableFrom(stringResolvableTypeMap.get(input).getRawClass())) {
if (KStream.class.isAssignableFrom(Objects.requireNonNull(stringResolvableTypeMap.get(input).getRawClass()))) {
final Class<?> valueClass =
(stringResolvableTypeMap.get(input).getGeneric(1).getRawClass() != null)
? (stringResolvableTypeMap.get(input).getGeneric(1).getRawClass()) : Object.class;
@@ -566,7 +569,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
public void setBeanFactory(@NonNull BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
}

98
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
View File

@@ -18,20 +18,22 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.api.RecordMetadata;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
import org.springframework.messaging.Message;
@@ -104,7 +106,7 @@ public class KafkaStreamsMessageConversionDelegate {
MessageHeaders messageHeaders = new MessageHeaders(headers);
final Message<?> convertedMessage = messageConverter.toMessage(message.getPayload(), messageHeaders);
perRecordContentTypeHolder.setContentType((String) messageHeaders.get(MessageHeaders.CONTENT_TYPE));
return convertedMessage.getPayload();
return Objects.requireNonNull(convertedMessage).getPayload();
});
kStreamWithEnrichedHeaders.process(() -> new Processor() {
@@ -112,19 +114,19 @@ public class KafkaStreamsMessageConversionDelegate {
ProcessorContext context;
@Override
public void init(ProcessorContext context) {
public void init(org.apache.kafka.streams.processor.api.ProcessorContext context) {
this.context = context;
}
@Override
public void process(Object key, Object value) {
public void process(Record record) {
if (perRecordContentTypeHolder.contentType != null) {
this.context.headers().remove(MessageHeaders.CONTENT_TYPE);
record.headers().remove(MessageHeaders.CONTENT_TYPE);
final Header header;
try {
header = new RecordHeader(MessageHeaders.CONTENT_TYPE,
new ObjectMapper().writeValueAsBytes(perRecordContentTypeHolder.contentType));
this.context.headers().add(header);
new ObjectMapper().writeValueAsBytes(perRecordContentTypeHolder.contentType));
record.headers().add(header);
}
catch (Exception e) {
if (LOG.isDebugEnabled()) {
@@ -159,7 +161,7 @@ public class KafkaStreamsMessageConversionDelegate {
resolvePerRecordContentType(bindingTarget, perRecordContentTypeHolder);
// Deserialize using a branching strategy
KStream<?, ?>[] branch = bindingTarget.branch(
Map<String, ? extends KStream<?, ?>> branchGraph = bindingTarget.split().branch(
// First filter where the message is converted and return true if
// everything went well, return false otherwise.
(o, o2) -> {
@@ -170,24 +172,24 @@ public class KafkaStreamsMessageConversionDelegate {
// further.
if (o2 != null) {
if (o2 instanceof Message || o2 instanceof String
|| o2 instanceof byte[]) {
|| o2 instanceof byte[]) {
Message<?> m1 = null;
if (o2 instanceof Message message) {
m1 = perRecordContentTypeHolder.contentType != null
? MessageBuilder.fromMessage(message)
.setHeader(
MessageHeaders.CONTENT_TYPE,
perRecordContentTypeHolder.contentType)
.build()
: message;
? MessageBuilder.fromMessage(message)
.setHeader(
MessageHeaders.CONTENT_TYPE,
perRecordContentTypeHolder.contentType)
.build()
: message;
}
else {
m1 = perRecordContentTypeHolder.contentType != null
? MessageBuilder.withPayload(o2).setHeader(
MessageHeaders.CONTENT_TYPE,
perRecordContentTypeHolder.contentType)
.build()
: MessageBuilder.withPayload(o2).build();
? MessageBuilder.withPayload(o2).setHeader(
MessageHeaders.CONTENT_TYPE,
perRecordContentTypeHolder.contentType)
.build()
: MessageBuilder.withPayload(o2).build();
}
convertAndSetMessage(o, valueClass, messageConverter, m1);
}
@@ -198,21 +200,23 @@ public class KafkaStreamsMessageConversionDelegate {
}
else {
LOG.info(
"Received a tombstone record. This will be skipped from further processing.");
"Received a tombstone record. This will be skipped from further processing.");
}
}
catch (Exception e) {
LOG.warn(
"Deserialization has failed. This will be skipped from further processing.",
e);
"Deserialization has failed. This will be skipped from further processing.",
e);
// pass through
failedWithDeserException[0] = e;
}
return isValidRecord;
},
}).branch(
// second filter that catches any messages for which an exception thrown
// in the first filter above.
(k, v) -> true);
(k, v) -> true)
.noDefaultBranch();
final KStream<?, ?>[] branch = branchGraph.values().toArray(new KStream[0]);
// process errors from the second filter in the branch above.
processErrorFromDeserialization(bindingTarget, branch[1], failedWithDeserException);
@@ -238,16 +242,15 @@ public class KafkaStreamsMessageConversionDelegate {
}
@Override
public void process(Object key, Object value) {
final Headers headers = this.context.headers();
final Iterable<Header> contentTypes = headers
.headers(MessageHeaders.CONTENT_TYPE);
public void process(Record record) {
final Iterable<Header> contentTypes = record.headers()
.headers(MessageHeaders.CONTENT_TYPE);
if (contentTypes != null && contentTypes.iterator().hasNext()) {
final String contentType = new String(
contentTypes.iterator().next().value());
contentTypes.iterator().next().value());
// remove leading and trailing quotes
final String cleanContentType = StringUtils.replace(contentType, "\"",
"");
"");
perRecordContentTypeHolder.setContentType(cleanContentType);
}
}
@@ -281,11 +284,14 @@ public class KafkaStreamsMessageConversionDelegate {
}
@Override
public void process(Object o, Object o2) {
public void process(Record record) {
// Only continue if the record was not a tombstone.
Object o = record.key();
Object o2 = record.value();
if (o2 != null) {
if (KafkaStreamsMessageConversionDelegate.this.kstreamBindingInformationCatalogue
.isDlqEnabled(bindingTarget)) {
.isDlqEnabled(bindingTarget)) {
if (o2 instanceof Message message) {
// We need to convert the key to a byte[] before sending to DLQ.
@@ -293,30 +299,34 @@ public class KafkaStreamsMessageConversionDelegate {
Serializer keySerializer = keySerde.serializer();
byte[] keyBytes = keySerializer.serialize(null, o);
ConsumerRecord consumerRecord = new ConsumerRecord(this.context.topic(), this.context.partition(), this.context.offset(),
if (this.context.recordMetadata().isPresent()) {
RecordMetadata recordMetadata = this.context.recordMetadata().get();
ConsumerRecord consumerRecord = new ConsumerRecord(recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset(),
keyBytes, message.getPayload());
KafkaStreamsMessageConversionDelegate.this.sendToDlqAndContinue
KafkaStreamsMessageConversionDelegate.this.sendToDlqAndContinue
.sendToDlq(consumerRecord, exception[0]);
}
}
else {
ConsumerRecord consumerRecord = new ConsumerRecord(this.context.topic(), this.context.partition(), this.context.offset(),
o, o2);
RecordMetadata recordMetadata = this.context.recordMetadata().get();
ConsumerRecord consumerRecord = new ConsumerRecord(recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset(),
o, o2);
KafkaStreamsMessageConversionDelegate.this.sendToDlqAndContinue
.sendToDlq(consumerRecord, exception[0]);
.sendToDlq(consumerRecord, exception[0]);
}
}
else if (KafkaStreamsMessageConversionDelegate.this.kstreamBinderConfigurationProperties
.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) {
.getDeserializationExceptionHandler() == DeserializationExceptionHandler.logAndFail) {
throw new IllegalStateException("Inbound deserialization failed. "
+ "Stopping further processing of records.");
+ "Stopping further processing of records.");
}
else if (KafkaStreamsMessageConversionDelegate.this.kstreamBinderConfigurationProperties
.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) {
.getDeserializationExceptionHandler() == DeserializationExceptionHandler.logAndContinue) {
// quietly passing through. No action needed, this is similar to
// log and continue.
LOG.error(
"Inbound deserialization failed. Skipping this record and continuing.");
"Inbound deserialization failed. Skipping this record and continuing.");
}
}
}

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java
View File

@@ -20,6 +20,7 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -79,8 +80,8 @@ public class KafkaStreamsRegistry {
public StreamsBuilderFactoryBean streamsBuilderFactoryBean(String applicationId) {
final Optional<StreamsBuilderFactoryBean> first = this.streamsBuilderFactoryBeanMap.values()
.stream()
.filter(streamsBuilderFactoryBean -> streamsBuilderFactoryBean.isRunning() && streamsBuilderFactoryBean
.getStreamsConfiguration().getProperty(StreamsConfig.APPLICATION_ID_CONFIG)
.filter(streamsBuilderFactoryBean -> streamsBuilderFactoryBean.isRunning() && Objects.requireNonNull(streamsBuilderFactoryBean
.getStreamsConfiguration()).getProperty(StreamsConfig.APPLICATION_ID_CONFIG)
.equals(applicationId))
.findFirst();
return first.orElse(null);

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022-2022 the original author or authors.
* Copyright 2022-2023 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.
@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Objects;
import java.util.StringJoiner;
import org.apache.kafka.streams.KafkaStreams;
@@ -35,6 +36,7 @@ import org.springframework.util.ReflectionUtils;
* Starting in kafka-streams 3.1 the topology exists at 'KafkaStreams.topologyMetadata'.
*
* @author Chris Bono
* @author Soby Chacko
* @since 3.2.6
*/
class KafkaStreamsVersionAgnosticTopologyInfoFacade {
@@ -80,7 +82,7 @@ class KafkaStreamsVersionAgnosticTopologyInfoFacade {
if (this.sourceTopicsForStoreMethod != null) {
this.sourceTopicsForStoreMethod.setAccessible(true);
logger.info(() -> "Using " + methodDescription(this.sourceTopicsForStoreMethod));
logger.info(() -> "Using " + methodDescription(Objects.requireNonNull(this.sourceTopicsForStoreMethod)));
}
else {
logger.warn("Could not find 'topologyMetadata.sourceTopicsForStore' or 'internalTopologyBuilder.sourceTopicsForStore' " +
@@ -97,13 +99,14 @@ class KafkaStreamsVersionAgnosticTopologyInfoFacade {
* @return {@code true} if state store is available or {@code false} if the state store is
* not available or there was a problem reflecting on the topology info
*/
@SuppressWarnings("unchecked")
boolean streamsAppActuallyHasStore(KafkaStreams kafkaStreams, String storeName) {
if (this.sourceTopicsForStoreMethod == null) {
logger.warn("Unable to reason about state store because sourceTopicsForStore method was not found - returning false");
return false;
}
try {
Object topologyInfo = ReflectionUtils.getField(this.topologyInfoField, kafkaStreams);
Object topologyInfo = ReflectionUtils.getField(Objects.requireNonNull(this.topologyInfoField), kafkaStreams);
if (topologyInfo == null) {
logger.warn("Unable to reason about state store because topologyInfo field was null - returning false");
return false;

14
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2023 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.
@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
@@ -36,11 +37,11 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.lang.NonNull;
import org.springframework.util.StringUtils;
/**
* Resolver for key and value Serde.
*
* On the inbound, if native decoding is enabled, then any deserialization on the value is
* handled by Kafka. First, we look for any key/value Serde set on the binding itself, if
* that is not available then look at the common Serde set at the global level. If that
@@ -48,17 +49,12 @@ import org.springframework.util.StringUtils;
* the deserialization on value and ignore any Serde set for value and rely on the
* contentType provided. Keys are always deserialized at the broker.
*
*
* Same rules apply on the outbound. If native encoding is enabled, then value
* serialization is done at the broker using any binder level Serde for value, if not
* using common Serde, if not, then byte[]. If native encoding is disabled, then the
* binder will do serialization using a contentType. Keys are always serialized by the
* broker.
*
* For state store, use serdes class specified in
* {@link org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsStateStore}
* to create Serde accordingly.
*
* @author Soby Chacko
* @author Lei Chen
* @author Eduard Domínguez
@@ -266,7 +262,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
private boolean isResolvableKStreamArrayType(ResolvableType resolvableType) {
return resolvableType.isArray() &&
KStream.class.isAssignableFrom(resolvableType.getComponentType().getRawClass());
KStream.class.isAssignableFrom(Objects.requireNonNull(resolvableType.getComponentType().getRawClass()));
}
private boolean isResolvalbeKafkaStreamsType(ResolvableType resolvableType) {
@@ -316,7 +312,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
context = (ConfigurableApplicationContext) applicationContext;
}

2
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java
View File

@@ -38,7 +38,7 @@ public class SendToDlqAndContinue implements ConsumerRecordRecoverer {
* DLQ dispatcher per topic in the application context. The key here is not the actual
* DLQ topic but the incoming topic that caused the error.
*/
private Map<String, DeadLetterPublishingRecoverer> dlqDispatchers = new HashMap<>();
private final Map<String, DeadLetterPublishingRecoverer> dlqDispatchers = new HashMap<>();
/**
* For a given topic, send the key/value record to DLQ topic.

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java
View File

@@ -30,6 +30,7 @@ import org.springframework.kafka.KafkaException;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.streams.KafkaStreamsMicrometerListener;
import org.springframework.lang.NonNull;
/**
* Iterate through all {@link StreamsBuilderFactoryBean} in the application context and
@@ -75,11 +76,9 @@ public class StreamsBuilderFactoryManager implements SmartLifecycle {
}
@Override
public void stop(Runnable callback) {
public void stop(@NonNull Runnable callback) {
stop();
if (callback != null) {
callback.run();
}
callback.run();
}
@Override

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/endpoint/KafkaStreamsTopologyEndpoint.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.endpoint;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;
@@ -52,7 +53,7 @@ public class KafkaStreamsTopologyEndpoint {
final List<String> descs = new ArrayList<>();
streamsBuilderFactoryBeans.stream()
.forEach(streamsBuilderFactoryBean ->
descs.add(streamsBuilderFactoryBean.getTopology().describe().toString()));
descs.add(Objects.requireNonNull(streamsBuilderFactoryBean.getTopology()).describe().toString()));
return descs;
}
@@ -61,7 +62,7 @@ public class KafkaStreamsTopologyEndpoint {
if (StringUtils.hasText(applicationId)) {
final StreamsBuilderFactoryBean streamsBuilderFactoryBean = this.kafkaStreamsRegistry.streamsBuilderFactoryBean(applicationId);
if (streamsBuilderFactoryBean != null) {
return streamsBuilderFactoryBean.getTopology().describe().toString();
return Objects.requireNonNull(streamsBuilderFactoryBean.getTopology()).describe().toString();
}
else {
return NO_TOPOLOGY_FOUND_MSG;

4
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java
View File

@@ -20,6 +20,7 @@ import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -55,7 +56,6 @@ public class FunctionDetectorCondition extends SpringBootCondition {
private static final Log LOG = LogFactory.getLog(FunctionDetectorCondition.class);
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public ConditionOutcome getMatchOutcome(ConditionContext context, AnnotatedTypeMetadata metadata) {
if (context != null && context.getBeanFactory() != null) {
@@ -89,7 +89,7 @@ public class FunctionDetectorCondition extends SpringBootCondition {
for (String key : functionComponents) {
final Class<?> classObj = ClassUtils.resolveClassName(((AnnotatedBeanDefinition)
context.getBeanFactory().getBeanDefinition(key))
Objects.requireNonNull(context.getBeanFactory()).getBeanDefinition(key))
.getMetadata().getClassName(),
ClassUtils.getDefaultClassLoader());
try {

14
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java
View File

@@ -22,6 +22,7 @@ import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -44,6 +45,7 @@ import org.springframework.cloud.stream.binding.BoundTargetHolder;
import org.springframework.cloud.stream.function.FunctionConstants;
import org.springframework.cloud.stream.function.StreamFunctionProperties;
import org.springframework.core.ResolvableType;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -71,9 +73,9 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
@Autowired
private StreamFunctionProperties streamFunctionProperties;
private ResolvableType[] types;
private final ResolvableType[] types;
private Method method;
private final Method method;
private final String functionName;
@@ -93,7 +95,7 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
"'bindingTargetFactories' cannot be empty");
int resolvableTypeDepthCounter = 0;
boolean isKafkaStreamsType = this.types[0].getRawClass().isAssignableFrom(KStream.class) ||
boolean isKafkaStreamsType = Objects.requireNonNull(this.types[0].getRawClass()).isAssignableFrom(KStream.class) ||
this.types[0].getRawClass().isAssignableFrom(KTable.class) ||
this.types[0].getRawClass().isAssignableFrom(GlobalKTable.class);
ResolvableType argument = isKafkaStreamsType ? this.types[0] : this.types[0].getGeneric(resolvableTypeDepthCounter++);
@@ -138,8 +140,8 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
final int lastTypeIndex = this.types.length - 1;
if (this.types.length > 1 && this.types[lastTypeIndex] != null && this.types[lastTypeIndex].getRawClass() != null) {
if (this.types[lastTypeIndex].getRawClass().isAssignableFrom(Function.class) ||
this.types[lastTypeIndex].getRawClass().isAssignableFrom(Consumer.class)) {
if (Objects.requireNonNull(this.types[lastTypeIndex].getRawClass()).isAssignableFrom(Function.class) ||
Objects.requireNonNull(this.types[lastTypeIndex].getRawClass()).isAssignableFrom(Consumer.class)) {
outboundArgument = this.types[lastTypeIndex].getGeneric(1);
}
}
@@ -266,7 +268,7 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
public void setBeanFactory(@NonNull BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}

3
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java
View File

@@ -51,6 +51,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderUtils;
import org.springframework.cloud.stream.function.StreamFunctionProperties;
import org.springframework.core.ResolvableType;
import org.springframework.lang.NonNull;
import org.springframework.util.ClassUtils;
/**
@@ -273,7 +274,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
public void setBeanFactory(@NonNull BeanFactory beanFactory) throws BeansException {
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
}
}

45
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java
View File

@@ -36,37 +36,12 @@ public class KafkaStreamsBinderConfigurationProperties
super(kafkaProperties);
}
/**
* Enumeration for various Serde errors.
*
* @deprecated in favor of {@link DeserializationExceptionHandler}.
*/
@Deprecated
public enum SerdeError {
/**
* Deserialization error handler with log and continue.
*/
logAndContinue,
/**
* Deserialization error handler with log and fail.
*/
logAndFail,
/**
* Deserialization error handler with DLQ send.
*/
sendToDlq
}
private String applicationId;
private StateStoreRetry stateStoreRetry = new StateStoreRetry();
private Map<String, Functions> functions = new HashMap<>();
private KafkaStreamsBinderConfigurationProperties.SerdeError serdeError;
/**
* {@link org.apache.kafka.streams.errors.DeserializationExceptionHandler} to use when
* there is a deserialization exception. This handler will be applied against all input bindings
@@ -100,26 +75,6 @@ public class KafkaStreamsBinderConfigurationProperties
this.applicationId = applicationId;
}
@Deprecated
public KafkaStreamsBinderConfigurationProperties.SerdeError getSerdeError() {
return this.serdeError;
}
@Deprecated
public void setSerdeError(
KafkaStreamsBinderConfigurationProperties.SerdeError serdeError) {
this.serdeError = serdeError;
if (serdeError == SerdeError.logAndContinue) {
this.deserializationExceptionHandler = DeserializationExceptionHandler.logAndContinue;
}
else if (serdeError == SerdeError.logAndFail) {
this.deserializationExceptionHandler = DeserializationExceptionHandler.logAndFail;
}
else if (serdeError == SerdeError.sendToDlq) {
this.deserializationExceptionHandler = DeserializationExceptionHandler.sendToDlq;
}
}
public DeserializationExceptionHandler getDeserializationExceptionHandler() {
return deserializationExceptionHandler;
}

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 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.
@@ -357,8 +357,7 @@ public class StreamToTableJoinFunctionTests {
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000",
"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.application-id" +
"=StreamToTableJoinFunctionTests-foobar",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(),
"--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) {
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
Thread.sleep(1000L);
// Input 2: Region per user (multiple records allowed per user).

2
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java
View File

@@ -70,7 +70,7 @@ public class DlqDestinationResolverTests {
"--spring.cloud.stream.bindings.process-in-0.destination=word1,word2",
"--spring.cloud.stream.bindings.process-out-0.destination=test-output",
"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.application-id=dlq-dest-resolver-test",
"--spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq",
"--spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler=sendToDlq",
"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.valueSerde="
+ "org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {

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

@@ -142,7 +142,6 @@ import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.core.BrokerAddress;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -271,14 +270,7 @@ public class KafkaBinderTests extends
private KafkaBinderConfigurationProperties createConfigurationProperties() {
var binderConfiguration = new KafkaBinderConfigurationProperties(
new TestKafkaProperties());
BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();
List<String> bAddresses = new ArrayList<>();
for (var bAddress : brokerAddresses) {
bAddresses.add(bAddress.toString());
}
var foo = new String[bAddresses.size()];
binderConfiguration.setBrokers(bAddresses.toArray(foo));
binderConfiguration.setBrokers(embeddedKafka.getBrokersAsString());
return binderConfiguration;
}
@@ -306,16 +298,9 @@ public class KafkaBinderTests extends
timeoutMultiplier = Double.parseDouble(multiplier);
}
BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();
var bAddresses = new ArrayList<>();
for (var bAddress : brokerAddresses) {
bAddresses.add(bAddress.toString());
}
var foo = new String[bAddresses.size()];
Map<String, Object> adminConfigs = new HashMap<>();
adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
bAddresses.toArray(foo)[0]);
embeddedKafka.getBrokersAsString());
adminClient = AdminClient.create(adminConfigs);
}

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

@@ -19,9 +19,11 @@ package org.springframework.cloud.stream.binder.kafka;
import java.util.Map;
import java.util.function.Function;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import reactor.core.publisher.Flux;
@@ -52,12 +54,19 @@ import static org.assertj.core.api.Assertions.assertThat;
*/
@ExtendWith(SpringExtension.class)
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
@EmbeddedKafka(topics = { "odd-topic", "even-topic" })
@EmbeddedKafka
class MultipleOutputBindingsPartitionsTests {
@Autowired
private EmbeddedKafkaBroker embeddedKafka;
@BeforeEach
public void before() {
NewTopic newTopic1 = new NewTopic("odd-topic", 10, (short) 1);
NewTopic newTopic2 = new NewTopic("even-topic", 5, (short) 1);
embeddedKafka.addTopics(newTopic1, newTopic2);
}
@Test
void singleInputTupleOutputWithDifferentPartitions() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(MultiOutputApplication.class)