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

Kafka Streams Binder Default Package Beans

Browse Source 
Currently, in Kafka Streams binder-based apps, processor beans
need to be declared public. This is unnecessary and caused by
some restrictions in the binder. This PR fixes this restriction.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2516
This commit is contained in:
Soby Chacko
2022-09-29 17:50:31 -04:00
parent 3fa4b67703
commit b85ee57d80
3 changed files with 6 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -93,7 +93,7 @@ public class FunctionDetectorCondition extends SpringBootCondition {
ClassUtils.getDefaultClassLoader());
try {
Method[] methods = classObj.getMethods();
Method[] methods = classObj.getDeclaredMethods();
Optional<Method> kafkaStreamMethod = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst();
// check if the bean name is overridden.
if (kafkaStreamMethod.isEmpty()) {

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

@@ -175,7 +175,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
.getMetadata().getClassName(),
ClassUtils.getDefaultClassLoader());
try {
Method[] methods = classObj.getMethods();
Method[] methods = classObj.getDeclaredMethods();
Optional<Method> functionalBeanMethods = Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst();
if (functionalBeanMethods.isEmpty()) {
final BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(key);

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

@@ -391,7 +391,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
InteractiveQueryService interactiveQueryService;
@Bean
public Function<KStream<Object, String>, KStream<String, WordCount>> process() {
Function<KStream<Object, String>, KStream<String, WordCount>> process() {
return input -> input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
@@ -405,7 +405,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
}
@Bean
public StreamsBuilderFactoryBeanConfigurer customizer() {
StreamsBuilderFactoryBeanConfigurer customizer() {
return fb -> {
try {
fb.setStateListener((newState, oldState) -> {
@@ -422,7 +422,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
}
@Bean
public StreamPartitioner<String, WordCount> streamPartitioner() {
StreamPartitioner<String, WordCount> streamPartitioner() {
return (t, k, v, n) -> k.equals("foo") ? 0 : 1;
}
}
@@ -431,7 +431,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
static class OutboundNullApplication {
@Bean
public Function<KStream<Object, String>, KStream<?, WordCount>> process() {
Function<KStream<Object, String>, KStream<?, WordCount>> process() {
return input -> input
.flatMapValues(
value -> Arrays.asList(value.toLowerCase().split("\\W+")))