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

GH-521: Fix pollable source client id

Browse Source 
Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/521

Previously, all pollable message sources got the client id `message.source`.
MBean registration failed with a warning when multiple pollable sources were present.

Use the binding name as the client id by default, overridable using the `client.id`
consumer property.

**cherry-pick to 2.0.x**

Resolves #523
This commit is contained in:
Gary Russell
2019-01-02 11:24:37 -05:00
committed by Oleg Zhurakousky
parent c51d7e0613
commit d63f9e5fa6
2 changed files with 16 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2018 the original author or authors.
* Copyright 2014-2019 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.
@@ -623,6 +623,7 @@ public class KafkaMessageChannelBinder extends
@Override
protected PolledConsumerResources createPolledConsumerResources(String name, String group,
ConsumerDestination destination, ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
boolean anonymous = !StringUtils.hasText(group);
Assert.isTrue(!anonymous || !consumerProperties.getExtension().isEnableDlq(),
"DLQ support is not available for anonymous subscriptions");
@@ -638,6 +639,11 @@ public class KafkaMessageChannelBinder extends
KafkaMessageSource<?, ?> source = new KafkaMessageSource<>(consumerFactory, topics);
source.setMessageConverter(getMessageConverter(consumerProperties));
source.setRawMessageHeader(consumerProperties.getExtension().isEnableDlq());
String clientId = name;
if (consumerProperties.getExtension().getConfiguration().containsKey(ConsumerConfig.CLIENT_ID_CONFIG)) {
clientId = consumerProperties.getExtension().getConfiguration().get(ConsumerConfig.CLIENT_ID_CONFIG);
}
source.setClientId(clientId);
if (!consumerProperties.isMultiplex()) {
// I copied this from the regular consumer - it looks bogus to me - includes all partitions

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2019 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.
@@ -35,6 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
@@ -2576,7 +2577,14 @@ public class KafkaBinderTests extends
Thread.sleep(100);
}
assertThat(polled).isTrue();
// Bind a second pollable consumer GH-521
consumerProps.getExtension().getConfiguration().put(ConsumerConfig.CLIENT_ID_CONFIG, "pollable2");
PollableSource<MessageHandler> second = new DefaultPollableMessageSource(this.messageConverter);
Binding<PollableSource<MessageHandler>> binding2 = binder.bindPollableConsumer("pollable2",
"group-polledConsumer2", second, consumerProps);
second.poll(m -> { });
binding.unbind();
binding2.unbind();
}
@SuppressWarnings({ "rawtypes", "unchecked" })