diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java index 7269875da..44bd7da0c 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java @@ -94,7 +94,7 @@ public class GlobalKTableBinder extends this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget), this.kafkaStreamsBindingInformationCatalogue, streamsBuilderFactoryBean); - return new DefaultBinding>(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); } } }; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java index 2065ee3c2..6bdc4e943 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java @@ -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 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; } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index e36221964..f2cc083e5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -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); } } }; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java index a5dcd9c09..0b8622787 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java @@ -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> doBindConsumer(String name, String group, KTable inputTarget, // @checkstyle:off @@ -95,7 +97,7 @@ class KTableBinder extends this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget), this.kafkaStreamsBindingInformationCatalogue, streamsBuilderFactoryBean); - return new DefaultBinding>(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); } } }; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java index 93eb00cf1..301246860 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java @@ -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; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java index 8cdb0bd78..57cc36701 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java @@ -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 metrics = kafkaStreams.metrics(); + final Map metrics = Objects.requireNonNull(kafkaStreams).metrics(); prepareToBindMetrics(registry, metrics); checkAndBindMetrics(registry, metrics); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 430e2e7c9..9c8efa213 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -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 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); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index 9a7f56c11..df2bdfaa5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -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, diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index 1511ee954..985a7ae71 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -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; } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java index 60cf9a5b8..509b42025 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java @@ -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> 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
contentTypes = headers - .headers(MessageHeaders.CONTENT_TYPE); + public void process(Record record) { + final Iterable
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."); } } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java index ced9ac5ca..839379fca 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java @@ -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 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); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java index 763030bc5..fc0fea376 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java @@ -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; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java index 3d73fd2f6..996b862ee 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java @@ -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; } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java index 569ef8c4f..2e048e96a 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java @@ -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 dlqDispatchers = new HashMap<>(); + private final Map dlqDispatchers = new HashMap<>(); /** * For a given topic, send the key/value record to DLQ topic. diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java index c0fa13e49..3798345a2 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java @@ -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 diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/endpoint/KafkaStreamsTopologyEndpoint.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/endpoint/KafkaStreamsTopologyEndpoint.java index e3e470bdb..8667f057f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/endpoint/KafkaStreamsTopologyEndpoint.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/endpoint/KafkaStreamsTopologyEndpoint.java @@ -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 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; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java index fc10220a3..16a8a92c9 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java @@ -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 { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java index 791457c24..55dfcf1ba 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java @@ -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; } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java index 8276eaecd..1c4b7c963 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java @@ -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; } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java index 95b1c93bc..9698b0bc5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java @@ -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 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; } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java index 6fba778d8..80546f136 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java @@ -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). diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java index 5b916ecc0..2d27f784e 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java @@ -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())) { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index efd506055..38487aef5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -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 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 adminConfigs = new HashMap<>(); adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, - bAddresses.toArray(foo)[0]); + embeddedKafka.getBrokersAsString()); adminClient = AdminClient.create(adminConfigs); } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/MultipleOutputBindingsPartitionsTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/MultipleOutputBindingsPartitionsTests.java index 982ce0536..00d52110e 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/MultipleOutputBindingsPartitionsTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/MultipleOutputBindingsPartitionsTests.java @@ -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)