diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index 146c2e88..da81b5d9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -22,6 +22,7 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.common.TopicPartition; @@ -30,8 +31,10 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; +import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * The base implementation for the {@link MessageListenerContainer}. @@ -104,6 +107,8 @@ public abstract class AbstractMessageListenerContainer } + protected final ConsumerFactory consumerFactory; // NOSONAR (final) + private final ContainerProperties containerProperties; private final Object lifecycleMonitor = new Object(); @@ -120,9 +125,27 @@ public abstract class AbstractMessageListenerContainer private volatile boolean paused; + /** + * Construct an instance with the provided properties. + * @param containerProperties the properties. + * @deprecated in favor of + * {@link #AbstractMessageListenerContainer(ConsumerFactory, ContainerProperties)}. + */ + @Deprecated protected AbstractMessageListenerContainer(ContainerProperties containerProperties) { - Assert.notNull(containerProperties, "'containerProperties' cannot be null"); + this(null, containerProperties); + } + /** + * Construct an instance with the provided factory and properties. + * @param consumerFactory the factory. + * @param containerProperties the properties. + */ + protected AbstractMessageListenerContainer(ConsumerFactory consumerFactory, + ContainerProperties containerProperties) { + + Assert.notNull(containerProperties, "'containerProperties' cannot be null"); + this.consumerFactory = consumerFactory; if (containerProperties.getTopics() != null) { this.containerProperties = new ContainerProperties(containerProperties.getTopics()); } @@ -219,6 +242,7 @@ public abstract class AbstractMessageListenerContainer @Override public final void start() { + checkGroupId(); synchronized (this.lifecycleMonitor) { if (!isRunning()) { Assert.isTrue( @@ -229,6 +253,21 @@ public abstract class AbstractMessageListenerContainer } } + public void checkGroupId() { + if (this.containerProperties.getTopicPartitions() == null) { + boolean hasGroupIdConsumerConfig = true; // assume true for non-standard containers + if (this.consumerFactory != null) { // we always have one for standard containers + Object groupIdConfig = this.consumerFactory.getConfigurationProperties() + .get(ConsumerConfig.GROUP_ID_CONFIG); + hasGroupIdConsumerConfig = groupIdConfig != null && groupIdConfig instanceof String + && StringUtils.hasText((String) groupIdConfig); + } + Assert.state(hasGroupIdConsumerConfig || StringUtils.hasText(this.containerProperties.getGroupId()), + "No group.id found in consumer config, container properties, or @KafkaListener annotation; " + + "a group.id is required when group management is used."); + } + } + protected abstract void doStart(); @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 e702599f..e38d6a7c 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 @@ -55,8 +55,6 @@ import org.springframework.util.Assert; */ public class ConcurrentMessageListenerContainer extends AbstractMessageListenerContainer { - private final ConsumerFactory consumerFactory; - private final List> containers = new ArrayList<>(); private int concurrency = 1; @@ -70,9 +68,8 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis */ public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { - super(containerProperties); + super(consumerFactory, containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); - this.consumerFactory = consumerFactory; } public int getConcurrency() { 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 15d67998..1db61d83 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 @@ -102,8 +102,6 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final AbstractMessageListenerContainer container; - private final ConsumerFactory consumerFactory; - private final TopicPartitionInitialOffset[] topicPartitions; private volatile ListenerConsumer listenerConsumer; @@ -159,10 +157,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener KafkaMessageListenerContainer(AbstractMessageListenerContainer container, ConsumerFactory consumerFactory, ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) { - super(containerProperties); + super(consumerFactory, containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); this.container = container == null ? this : container; - this.consumerFactory = consumerFactory; if (topicPartitions != null) { this.topicPartitions = Arrays.copyOf(topicPartitions, topicPartitions.length); } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 7cd19761..95247568 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -19,7 +19,6 @@ package org.springframework.kafka.listener; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; @@ -395,7 +394,7 @@ public class ConcurrentMessageListenerContainerTests { }; ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), anyString(), anyString())).willReturn(consumer); + given(cf.createConsumer(anyString(), anyString(), anyString())).willReturn(consumer); given(consumer.poll(anyLong())) .willAnswer(new Answer>() { @@ -407,6 +406,7 @@ public class ConcurrentMessageListenerContainerTests { }); ContainerProperties containerProps = new ContainerProperties(topic1PartitionS); + containerProps.setGroupId("grp"); containerProps.setMessageListener((MessageListener) message -> { }); ConcurrentMessageListenerContainer container = diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 74d3a677..7576a4ee 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -499,7 +499,7 @@ public class KafkaMessageListenerContainerTests { public void testRecordAckMock() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer); + given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer); final Map>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), @@ -512,6 +512,7 @@ public class KafkaMessageListenerContainerTests { TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] { new TopicPartitionInitialOffset("foo", 0) }; ContainerProperties containerProps = new ContainerProperties(topicPartition); + containerProps.setGroupId("grp"); containerProps.setAckMode(AckMode.RECORD); final CountDownLatch latch = new CountDownLatch(2); MessageListener messageListener = spy( @@ -565,7 +566,7 @@ public class KafkaMessageListenerContainerTests { private void testRecordAckMockForeignThreadGuts(AckMode ackMode) throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer); + given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer); final Map>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), @@ -578,6 +579,7 @@ public class KafkaMessageListenerContainerTests { TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] { new TopicPartitionInitialOffset("foo", 0) }; ContainerProperties containerProps = new ContainerProperties(topicPartition); + containerProps.setGroupId("grp"); containerProps.setAckMode(ackMode); final CountDownLatch latch = new CountDownLatch(2); final List acks = new ArrayList<>(); @@ -627,7 +629,7 @@ public class KafkaMessageListenerContainerTests { public void testNonResponsiveConsumerEvent() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), eq(""), isNull())).willReturn(consumer); + given(cf.createConsumer(eq("grp"), eq(""), isNull())).willReturn(consumer); final Map>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), @@ -644,6 +646,7 @@ public class KafkaMessageListenerContainerTests { TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] { new TopicPartitionInitialOffset("foo", 0) }; ContainerProperties containerProps = new ContainerProperties(topicPartition); + containerProps.setGroupId("grp"); containerProps.setNoPollThreshold(2.0f); containerProps.setPollTimeout(10); containerProps.setMonitorInterval(1); @@ -1703,7 +1706,7 @@ public class KafkaMessageListenerContainerTests { public void testPauseResume() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer); + given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer); final Map>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), @@ -1735,6 +1738,7 @@ public class KafkaMessageListenerContainerTests { TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] { new TopicPartitionInitialOffset("foo", 0) }; ContainerProperties containerProps = new ContainerProperties(topicPartition); + containerProps.setGroupId("grp"); containerProps.setAckMode(AckMode.RECORD); containerProps.setClientId("clientId"); containerProps.setIdleEventInterval(100L); @@ -1764,7 +1768,7 @@ public class KafkaMessageListenerContainerTests { public void testInitialSeek() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer); + given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer); ConsumerRecords emptyRecords = new ConsumerRecords<>(Collections.emptyMap()); final CountDownLatch latch = new CountDownLatch(1); given(consumer.poll(anyLong())).willAnswer(i -> { @@ -1781,6 +1785,7 @@ public class KafkaMessageListenerContainerTests { new TopicPartitionInitialOffset("foo", 5, SeekPosition.END), }; ContainerProperties containerProps = new ContainerProperties(topicPartition); + containerProps.setGroupId("grp"); containerProps.setAckMode(AckMode.RECORD); containerProps.setClientId("clientId"); containerProps.setMessageListener((MessageListener) r -> { }); @@ -1868,7 +1873,7 @@ public class KafkaMessageListenerContainerTests { public void testAckModeCount() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer); + given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer); TopicPartition topicPartition = new TopicPartition("foo", 0); final Map>> records1 = new HashMap<>(); records1.put(topicPartition, Arrays.asList( @@ -1911,6 +1916,7 @@ public class KafkaMessageListenerContainerTests { TopicPartitionInitialOffset[] topicPartitionOffset = new TopicPartitionInitialOffset[] { new TopicPartitionInitialOffset("foo", 0) }; ContainerProperties containerProps = new ContainerProperties(topicPartitionOffset); + containerProps.setGroupId("grp"); containerProps.setAckMode(AckMode.COUNT); containerProps.setAckCount(3); containerProps.setClientId("clientId"); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java new file mode 100644 index 00000000..9b6fbd67 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java @@ -0,0 +1,142 @@ +/* + * Copyright 2018 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.listener; + +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; + +import java.util.Collections; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.context.ApplicationContextException; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.TopicPartitionInitialOffset; +import org.springframework.kafka.test.rule.KafkaEmbedded; + +/** + * @author Gary Russell + * @since 2.1.5 + * + */ +public class MissingGroupIdTests { + + @ClassRule + public static KafkaEmbedded kafkaEmbedded = new KafkaEmbedded(1, true, "missing.group"); + + @Test + public void testContextFailsWithKafkaListener() { + assertThatExceptionOfType(ApplicationContextException.class).isThrownBy(() -> { + new AnnotationConfigApplicationContext(Config1.class); + }) + .withCauseInstanceOf(IllegalStateException.class) + .withMessageContaining("No group.id found in consumer config"); + } + + @Test + public void testContextFailsWithSubscribedContainer() { + assertThatExceptionOfType(ApplicationContextException.class).isThrownBy(() -> { + new AnnotationConfigApplicationContext(Config2.class); + }) + .withCauseInstanceOf(IllegalStateException.class) + .withMessageContaining("No group.id found in consumer config"); + } + + @Test + public void testContextLoadsWithAssignedContainer() { + new AnnotationConfigApplicationContext(Config3.class).close(); + } + + @Configuration + @EnableKafka + public static class Config1 { + + @Bean + public ConsumerFactory cf() { + return new DefaultKafkaConsumerFactory<>( + Collections.singletonMap(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + kafkaEmbedded.getBrokersAsString()), + new StringDeserializer(), new StringDeserializer()); + } + + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(cf()); + return factory; + } + + @KafkaListener(topics = "missing.group") + public void listen(String in) { + // no op + } + + } + + @Configuration + @EnableKafka + public static class Config2 { + + @Bean + public ConsumerFactory cf() { + return new DefaultKafkaConsumerFactory<>( + Collections.singletonMap(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + kafkaEmbedded.getBrokersAsString()), + new StringDeserializer(), new StringDeserializer()); + } + + @Bean + public KafkaMessageListenerContainer container() { + ContainerProperties props = new ContainerProperties("missing.group"); + return new KafkaMessageListenerContainer<>(cf(), props); + } + + } + + @Configuration + @EnableKafka + public static class Config3 { + + @Bean + public ConsumerFactory cf() { + return new DefaultKafkaConsumerFactory<>( + Collections.singletonMap(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + kafkaEmbedded.getBrokersAsString()), + new StringDeserializer(), new StringDeserializer()); + } + + @Bean + public KafkaMessageListenerContainer container() { + ContainerProperties props = new ContainerProperties(new TopicPartitionInitialOffset("missing.group", 0)); + props.setMessageListener((MessageListener) r -> { }); + return new KafkaMessageListenerContainer<>(cf(), props); + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java index 90e05f55..6bd89a26 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java @@ -125,7 +125,7 @@ public class SeekToCurrentOnErrorBatchModeTests { private int count; - @KafkaListener(topics = "foo") + @KafkaListener(topics = "foo", groupId = "grp") public void foo(String in) { this.contents.add(in); this.deliveryLatch.countDown(); @@ -139,7 +139,7 @@ public class SeekToCurrentOnErrorBatchModeTests { public ConsumerFactory consumerFactory() { ConsumerFactory consumerFactory = mock(ConsumerFactory.class); final Consumer consumer = consumer(); - given(consumerFactory.createConsumer(null, "", "-0")).willReturn(consumer); + given(consumerFactory.createConsumer("grp", "", "-0")).willReturn(consumer); return consumerFactory; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java index 1585e5e6..b7dd8141 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java @@ -129,7 +129,7 @@ public class SeekToCurrentOnErrorRecordModeTests { private int count; - @KafkaListener(topics = "foo") + @KafkaListener(topics = "foo", groupId = "grp") public void foo(String in) { this.contents.add(in); this.deliveryLatch.countDown(); @@ -143,7 +143,7 @@ public class SeekToCurrentOnErrorRecordModeTests { public ConsumerFactory consumerFactory() { ConsumerFactory consumerFactory = mock(ConsumerFactory.class); final Consumer consumer = consumer(); - given(consumerFactory.createConsumer(null, "", "-0")).willReturn(consumer); + given(consumerFactory.createConsumer("grp", "", "-0")).willReturn(consumer); return consumerFactory; }