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

Remove duplicate KafkaStreams topology endpoint

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/895
This commit is contained in:
Soby Chacko
2020-10-22 12:30:47 -04:00
parent 5cdd8a09f9
commit 0a0d3a1057
3 changed files with 9 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/src/main/asciidoc/kafka-streams.adoc
View File

@@ -1468,7 +1468,7 @@ Kafka Streams binder provides the following actuator endpoints for retrieving th
`/actuator/kafkastreamstopology`
`/actuator/kafkastreamstopology/<applicaiton-id of the processor>`
`/actuator/kafkastreamstopology/<application-id of the processor>`
You need to include the actuator and web dependencies from Spring Boot to access these endpoints.
Further, you also need to add `kafkastreamstopology` to `management.endpoints.web.exposure.include` property.

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.kafka.streams.endpoint;
import java.util.ArrayList;
import java.util.List;
import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
@@ -46,13 +47,14 @@ public class KafkaStreamsTopologyEndpoint {
}
@ReadOperation
public String kafkaStreamsTopology() {
public List<String> kafkaStreamsTopologies() {
final List<StreamsBuilderFactoryBean> streamsBuilderFactoryBeans = this.kafkaStreamsRegistry.streamsBuilderFactoryBeans();
final StringBuilder topologyDescription = new StringBuilder();
final List<String> descs = new ArrayList<>();
streamsBuilderFactoryBeans.stream()
.forEach(streamsBuilderFactoryBean ->
topologyDescription.append(streamsBuilderFactoryBean.getTopology().describe().toString()));
return topologyDescription.toString();
descs.add(streamsBuilderFactoryBean.getTopology().describe().toString()));
return descs;
}
@ReadOperation

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

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.function;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -112,7 +113,8 @@ public class KafkaStreamsBinderWordCountFunctionTests {
//Testing topology endpoint
final KafkaStreamsRegistry kafkaStreamsRegistry = context.getBean(KafkaStreamsRegistry.class);
final KafkaStreamsTopologyEndpoint kafkaStreamsTopologyEndpoint = new KafkaStreamsTopologyEndpoint(kafkaStreamsRegistry);
final String topology1 = kafkaStreamsTopologyEndpoint.kafkaStreamsTopology();
final List<String> topologies = kafkaStreamsTopologyEndpoint.kafkaStreamsTopologies();
final String topology1 = topologies.get(0);
final String topology2 = kafkaStreamsTopologyEndpoint.kafkaStreamsTopology("testKstreamWordCountFunction");
assertThat(topology1).isNotEmpty();
assertThat(topology1).isEqualTo(topology2);