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

Code cleanup - Kafka Streams binder

Browse Source
This commit is contained in:
Soby Chacko
2022-07-11 14:33:18 -04:00
parent bc72fc7155
commit 89eb9a31d4
10 changed files with 33 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -129,16 +129,10 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
.getStartOffset();
Topology.AutoOffsetReset autoOffsetReset = null;
if (startOffset != null) {
switch (startOffset) {
case earliest:
autoOffsetReset = Topology.AutoOffsetReset.EARLIEST;
break;
case latest:
autoOffsetReset = Topology.AutoOffsetReset.LATEST;
break;
default:
break;
}
autoOffsetReset = switch (startOffset) {
case earliest -> Topology.AutoOffsetReset.EARLIEST;
case latest -> Topology.AutoOffsetReset.LATEST;
};
}
if (extendedConsumerProperties.isResetOffsets()) {
AbstractKafkaStreamsBinderProcessor.LOG.warn("Detected resetOffsets configured on binding "
@@ -221,11 +215,11 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
final MutablePropertySources propertySources = environment.getPropertySources();
if (!StringUtils.isEmpty(bindingProperties.getBinder())) {
if (StringUtils.hasText(bindingProperties.getBinder())) {
final KafkaStreamsBinderConfigurationProperties multiBinderKafkaStreamsBinderConfigurationProperties =
applicationContext.getBean(bindingProperties.getBinder() + "-KafkaStreamsBinderConfigurationProperties", KafkaStreamsBinderConfigurationProperties.class);
String connectionString = multiBinderKafkaStreamsBinderConfigurationProperties.getKafkaConnectionString();
if (StringUtils.isEmpty(connectionString)) {
if (!StringUtils.hasText(connectionString)) {
connectionString = (String) propertySources.get(bindingProperties.getBinder() + "-kafkaStreamsBinderEnv").getProperty("spring.cloud.stream.kafka.binder.brokers");
}
@@ -587,7 +581,10 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
// Processor to retrieve the header value.
stream.process(() -> eventTypeProcessor(kafkaStreamsConsumerProperties, matched, topicObject, headersObject));
// Branching based on event type match.
final KStream<?, ?>[] branch = stream.branch((key, value) -> matched.getAndSet(false));
final Map<String, ? extends KStream<?, ?>> stringKStreamMap = stream.split()
.branch((key, value) -> matched.getAndSet(false))
.noDefaultBranch();
final KStream<?, ?>[] branch = stringKStreamMap.values().toArray(new KStream[0]);
// Deserialize if we have a branch from above.
final KStream<?, Object> deserializedKStream = branch[0].mapValues(value -> valueSerde.deserializer().deserialize(
topicObject.get(), headersObject.get(), ((Bytes) value).get()));
@@ -600,7 +597,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
private <K, V> Consumed<K, V> getConsumed(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties,
Serde<K> keySerde, Serde<V> valueSerde, Topology.AutoOffsetReset autoOffsetReset) {
TimestampExtractor timestampExtractor = null;
if (!StringUtils.isEmpty(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName())) {
if (StringUtils.hasText(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName())) {
timestampExtractor = applicationContext.getBean(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName(),
TimestampExtractor.class);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2019 the original author or authors.
* Copyright 2019-2022 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.
@@ -37,7 +37,7 @@ public class EncodingDecodingBindAdviceHandler implements ConfigurationPropertie
private boolean decodingSettingProvided;
public boolean isDecodingSettingProvided() {
return decodingSettingProvided;
return this.decodingSettingProvided;
}
public boolean isEncodingSettingProvided() {
@@ -46,7 +46,7 @@ public class EncodingDecodingBindAdviceHandler implements ConfigurationPropertie
@Override
public BindHandler apply(BindHandler bindHandler) {
BindHandler handler = new AbstractBindHandler(bindHandler) {
return new AbstractBindHandler(bindHandler) {
@Override
public <T> Bindable<T> onStart(ConfigurationPropertyName name,
Bindable<T> target, BindContext context) {
@@ -67,6 +67,5 @@ public class EncodingDecodingBindAdviceHandler implements ConfigurationPropertie
return bindHandler.onStart(name, target, context);
}
};
return handler;
}
}

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

@@ -169,7 +169,7 @@ public class InteractiveQueryService {
String applicationServer = configuration.get("application.server");
String[] splits = StringUtils.split(applicationServer, ":");
return new HostInfo(splits[0], Integer.valueOf(splits[1]));
return new HostInfo(splits[0], Integer.parseInt(splits[1]));
}
return null;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2021 the original author or authors.
* Copyright 2017-2022 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.
@@ -244,7 +244,7 @@ class KStreamBinder extends
produced.withName(properties.getProducedAs());
}
StreamPartitioner streamPartitioner = null;
if (!StringUtils.isEmpty(properties.getStreamPartitionerBeanName())) {
if (StringUtils.hasText(properties.getStreamPartitionerBeanName())) {
streamPartitioner = getApplicationContext().getBean(properties.getStreamPartitionerBeanName(),
StreamPartitioner.class);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2021 the original author or authors.
* Copyright 2019-2022 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.
@@ -163,9 +163,9 @@ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator i
if (isRunningResult) {
final Set<ThreadMetadata> threadMetadata = kafkaStreams.metadataForLocalThreads();
final Map<String, Object> threadDetails = new HashMap();
final Map<String, Object> threadDetails = new HashMap<>();
for (ThreadMetadata metadata : threadMetadata) {
final Map<String, Object> threadDetail = new HashMap();
final Map<String, Object> threadDetail = new HashMap<>();
threadDetail.put("threadName", metadata.threadName());
threadDetail.put("threadState", metadata.threadState());
threadDetail.put("adminClientId", metadata.adminClientId());

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2021 the original author or authors.
* Copyright 2017-2022 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.
@@ -222,8 +222,7 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
kafkaConnectionString);
}
}
else if (bootstrapServerConfig instanceof List) {
List bootStrapCollection = (List) bootstrapServerConfig;
else if (bootstrapServerConfig instanceof List bootStrapCollection) {
if (bootStrapCollection.size() == 1 && bootStrapCollection.get(0).equals("localhost:9092")) {
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaConnectionString);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2021 the original author or authors.
* Copyright 2019-2022 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.
@@ -560,9 +560,6 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
throw new IllegalStateException(ex);
}
}
else {
//throw new IllegalStateException(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
return arguments;
}

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

@@ -215,8 +215,8 @@ abstract class SerdeResolverUtils {
* Private internal class used strictly to 'remember' a score for a serde and use it for sorting later.
*/
private static class SerdeWithSpecificityScore implements Comparable<SerdeWithSpecificityScore> {
private Integer score;
private String serdeBeanName;
private final Integer score;
private final String serdeBeanName;
SerdeWithSpecificityScore(Integer score, String serdeBeanName) {
this.score = Objects.requireNonNull(score);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2021 the original author or authors.
* Copyright 2019-2022 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.
@@ -96,7 +96,7 @@ public class FunctionDetectorCondition extends SpringBootCondition {
Method[] methods = classObj.getMethods();
Optional<Method> kafkaStreamMethod = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst();
// check if the bean name is overridden.
if (!kafkaStreamMethod.isPresent()) {
if (kafkaStreamMethod.isEmpty()) {
final BeanDefinition beanDefinition = context.getBeanFactory().getBeanDefinition(key);
final String factoryMethodName = beanDefinition.getFactoryMethodName();
kafkaStreamMethod = Arrays.stream(methods).filter(m -> m.getName().equals(factoryMethodName)).findFirst();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2021 the original author or authors.
* Copyright 2019-2022 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.
@@ -65,13 +65,13 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
private ConfigurableListableBeanFactory beanFactory;
private boolean onlySingleFunction;
private Map<String, ResolvableType> resolvableTypeMap = new TreeMap<>();
private Map<String, Method> methods = new TreeMap<>();
private final Map<String, ResolvableType> resolvableTypeMap = new TreeMap<>();
private final Map<String, Method> methods = new TreeMap<>();
private final StreamFunctionProperties streamFunctionProperties;
private Map<String, ResolvableType> kafkaStreamsOnlyResolvableTypes = new HashMap<>();
private Map<String, Method> kafakStreamsOnlyMethods = new HashMap<>();
private final Map<String, ResolvableType> kafkaStreamsOnlyResolvableTypes = new HashMap<>();
private final Map<String, Method> kafakStreamsOnlyMethods = new HashMap<>();
public KafkaStreamsFunctionBeanPostProcessor(StreamFunctionProperties streamFunctionProperties) {
this.streamFunctionProperties = streamFunctionProperties;
@@ -177,7 +177,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
try {
Method[] methods = classObj.getMethods();
Optional<Method> functionalBeanMethods = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst();
if (!functionalBeanMethods.isPresent()) {
if (functionalBeanMethods.isEmpty()) {
final BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(key);
final String factoryMethodName = beanDefinition.getFactoryMethodName();
functionalBeanMethods = Arrays.stream(methods).filter(m -> m.getName().equals(factoryMethodName)).findFirst();