From 40c1a8b8f117faee1c4342043baa7b186c96cdbf Mon Sep 17 00:00:00 2001 From: Alberto Manzaneque Date: Wed, 10 Oct 2018 15:50:06 +0200 Subject: [PATCH] Setting group in TopicInformation correctly for anonymous consumers. This makes metrics available for anonymous consumers as well. A couple of tests to verify that TopicInformation is built correctly Removing unneeded import --- .../binder/kafka/KafkaMessageChannelBinder.java | 4 ++-- .../stream/binder/kafka/KafkaBinderTests.java | 15 +++++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 926e19475..eb6b8a0e6 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -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())); } } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 29d0bc149..2befeb05f 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -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 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 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();