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

Setting group in TopicInformation correctly for anonymous consumers.

Browse Source 
This makes metrics available for anonymous consumers as well.

A couple of tests to verify that TopicInformation is built correctly

Removing unneeded import
This commit is contained in:
Alberto Manzaneque
2018-10-10 15:50:06 +02:00
committed by Soby Chacko
parent c43e45c7ad
commit 40c1a8b8f1
2 changed files with 17 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -380,12 +380,12 @@ public class KafkaMessageChannelBinder extends
+ "use the regex pattern to specify multiple topics instead");
boolean groupManagement = extendedConsumerProperties.getExtension().isAutoRebalanceEnabled();
if (!extendedConsumerProperties.isMultiplex()) {
listenedPartitions.addAll(processTopic(group, extendedConsumerProperties, consumerFactory,
listenedPartitions.addAll(processTopic(consumerGroup, extendedConsumerProperties, consumerFactory,
partitionCount, usingPatterns, groupManagement, destination.getName()));
}
else {
for (String name : StringUtils.commaDelimitedListToStringArray(destination.getName())) {
listenedPartitions.addAll(processTopic(group, extendedConsumerProperties, consumerFactory,
listenedPartitions.addAll(processTopic(consumerGroup, extendedConsumerProperties, consumerFactory,
partitionCount, usingPatterns, groupManagement, name.trim()));
}
}

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

@@ -79,6 +79,7 @@ import org.springframework.cloud.stream.binder.PollableSource;
import org.springframework.cloud.stream.binder.RequeueCurrentMessageException;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.binder.TestUtils;
import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.TopicInformation;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
@@ -412,6 +413,7 @@ public class KafkaBinderTests extends
.setHeader(MessageHeaders.CONTENT_TYPE,
MimeTypeUtils.APPLICATION_OCTET_STREAM)
.build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
moduleOutputChannel.send(message);
@@ -431,6 +433,13 @@ public class KafkaBinderTests extends
assertThat(new String(inboundMessageRef.get().getPayload(), StandardCharsets.UTF_8)).isEqualTo("foo");
assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE))
.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM);
Map<String, TopicInformation> topicsInUse = ((KafkaTestBinder)binder).getCoreBinder().getTopicsInUse();
assertThat(topicsInUse.keySet()).contains("foo.bar");
TopicInformation topic = topicsInUse.get("foo.bar");
assertThat(topic.isConsumerTopic()).isTrue();
assertThat(topic.getConsumerGroup()).isEqualTo("testSendAndReceive");
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -1457,6 +1466,12 @@ public class KafkaBinderTests extends
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload(), StandardCharsets.UTF_8)).isEqualTo(testPayload3);
Map<String, TopicInformation> topicsInUse = ((KafkaTestBinder)binder).getCoreBinder().getTopicsInUse();
assertThat(topicsInUse.keySet()).contains("defaultGroup.0");
TopicInformation topic = topicsInUse.get("defaultGroup.0");
assertThat(topic.isConsumerTopic()).isTrue();
assertThat(topic.getConsumerGroup()).startsWith("anonymous");
producerBinding.unbind();
binding1.unbind();
binding2.unbind();