diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/ClientIdSuffixAware.java b/spring-kafka/src/main/java/org/springframework/kafka/core/ClientIdSuffixAware.java new file mode 100644 index 00000000..61bb0567 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/ClientIdSuffixAware.java @@ -0,0 +1,45 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.core; + +import org.apache.kafka.clients.consumer.Consumer; + +/** + * If a {@code ConsumerFactory} implements this interface, it supports customizing the + * {@code client.id} by adding a suffix, if a {@code client.id} property is provided. + * + * @deprecated - in 2.0, this method will move to {@link ConsumerFactory}. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * @since 1.0.6 + * + */ +@Deprecated +public interface ClientIdSuffixAware { + + /** + * Create a consumer, appending the suffix to the {@code client.id} property, + * if present. + * @param clientIdSuffix the suffix. + * @return the consumer. + */ + Consumer createConsumer(String clientIdSuffix); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java index 466659b3..351081fe 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java @@ -16,6 +16,7 @@ package org.springframework.kafka.core; +import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -36,7 +37,8 @@ import org.apache.kafka.common.serialization.Deserializer; * @author Gary Russell * @author Murali Reddy */ -public class DefaultKafkaConsumerFactory implements ConsumerFactory { +@SuppressWarnings("deprecation") +public class DefaultKafkaConsumerFactory implements ConsumerFactory, ClientIdSuffixAware { private final Map configs; @@ -64,13 +66,44 @@ public class DefaultKafkaConsumerFactory implements ConsumerFactory this.valueDeserializer = valueDeserializer; } + /** + * Return an unmodifiable reference to the configuration map for this factory. + * Useful for cloning to make a similar factory. + * @return the configs. + * @since 1.0.6 + */ + public Map getConfigurationProperties() { + return Collections.unmodifiableMap(this.configs); + } + @Override public Consumer createConsumer() { return createKafkaConsumer(); } + @Override + public Consumer createConsumer(String clientIdSuffix) { + return createKafkaConsumer(clientIdSuffix); + } + protected KafkaConsumer createKafkaConsumer() { - return new KafkaConsumer(this.configs, this.keyDeserializer, this.valueDeserializer); + return createKafkaConsumer(this.configs); + } + + protected KafkaConsumer createKafkaConsumer(String clientIdSuffix) { + if (!this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG) || clientIdSuffix == null) { + return createKafkaConsumer(); + } + else { + Map modifiedClientIdConfigs = new HashMap<>(this.configs); + modifiedClientIdConfigs.put(ConsumerConfig.CLIENT_ID_CONFIG, + modifiedClientIdConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG) + clientIdSuffix); + return createKafkaConsumer(modifiedClientIdConfigs); + } + } + + protected KafkaConsumer createKafkaConsumer(Map configs) { + return new KafkaConsumer(configs, this.keyDeserializer, this.valueDeserializer); } @Override diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index f3ad5b11..f98ee548 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -123,6 +123,7 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis if (getApplicationEventPublisher() != null) { container.setApplicationEventPublisher(getApplicationEventPublisher()); } + container.setClientIdSuffix("-" + i); container.start(); this.containers.add(container); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index b28ce83a..56a8f94c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -87,6 +87,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private AcknowledgingMessageListener acknowledgingMessageListener; + private String clientIdSuffix; + /** * Construct an instance with the supplied configuration properties. * @param consumerFactory the consumer factory. @@ -117,6 +119,16 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } + /** + * Set a suffix to add to the {@code client.id} consumer property (if the consumer + * factory supports it). + * @param clientIdSuffix the suffix to add. + * @since 1.0.6 + */ + public void setClientIdSuffix(String clientIdSuffix) { + this.clientIdSuffix = clientIdSuffix; + } + /** * Return the {@link TopicPartition}s currently assigned to this container, * either explicitly or by Kafka; may be null if not assigned yet. @@ -214,7 +226,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener @Override public String toString() { - return "KafkaMessageListenerContainer [id=" + getBeanName() + ", topicPartitions=" + getAssignedPartitions() + return "KafkaMessageListenerContainer [id=" + getBeanName() + + (this.clientIdSuffix != null ? ", clientIndex=" + this.clientIdSuffix : "") + + ", topicPartitions=" + getAssignedPartitions() + "]"; } @@ -278,10 +292,17 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener */ private boolean paused; - private ListenerConsumer(MessageListener listener, AcknowledgingMessageListener ackListener) { + @SuppressWarnings({"deprecation", "unchecked"}) + ListenerConsumer(MessageListener listener, AcknowledgingMessageListener ackListener) { Assert.state(!this.isAnyManualAck || !this.autoCommit, "Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode()); - final Consumer consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer(); + final Consumer consumer = + KafkaMessageListenerContainer.this.consumerFactory + instanceof org.springframework.kafka.core.ClientIdSuffixAware + ? ((org.springframework.kafka.core.ClientIdSuffixAware) KafkaMessageListenerContainer + .this.consumerFactory) + .createConsumer(KafkaMessageListenerContainer.this.clientIdSuffix) + : KafkaMessageListenerContainer.this.consumerFactory.createConsumer(); ConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() { @@ -758,6 +779,10 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private volatile Thread executingThread; + ListenerInvoker() { + super(); + } + @Override public void run() { Assert.isTrue(this.active, "This instance is not active anymore"); @@ -831,7 +856,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final boolean immediate; - private ConsumerAcknowledgment(ConsumerRecord record, boolean immediate) { + ConsumerAcknowledgment(ConsumerRecord record, boolean immediate) { this.record = record; this.immediate = immediate; } @@ -866,6 +891,10 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private static final Log logger = LogFactory.getLog(LoggingCommitCallback.class); + LoggingCommitCallback() { + super(); + } + @Override public void onComplete(Map offsets, Exception exception) { if (exception != null) { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java new file mode 100644 index 00000000..b1cee31c --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java @@ -0,0 +1,50 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Collections; +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.junit.Test; + +/** + * @author Gary Russell + * @since 1.0.6 + * + */ +public class DefaultKafkaConsumerFactoryTests { + + @Test + public void testClientId() { + Map configs = Collections.singletonMap(ConsumerConfig.CLIENT_ID_CONFIG, "foo"); + DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(configs) { + + @Override + protected KafkaConsumer createKafkaConsumer(Map configs) { + assertThat(configs.get(ConsumerConfig.CLIENT_ID_CONFIG)).isEqualTo("foo-1"); + return null; + } + + }; + factory.createConsumer("-1"); + } + +}