Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-631: Fail fast with missing required group.id

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/631
This commit is contained in:
Gary Russell
2018-03-30 14:39:10 -04:00
committed by Artem Bilan
parent 506ba35783
commit 20635b8004
8 changed files with 202 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
View File

@@ -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<K, V>
}
protected final ConsumerFactory<K, V> consumerFactory; // NOSONAR (final)
private final ContainerProperties containerProperties;
private final Object lifecycleMonitor = new Object();
@@ -120,9 +125,27 @@ public abstract class AbstractMessageListenerContainer<K, V>
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<K, V> 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<K, V>
@Override
public final void start() {
checkGroupId();
synchronized (this.lifecycleMonitor) {
if (!isRunning()) {
Assert.isTrue(
@@ -229,6 +253,21 @@ public abstract class AbstractMessageListenerContainer<K, V>
}
}
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

5
spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
View File

@@ -55,8 +55,6 @@ import org.springframework.util.Assert;
*/
public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
private final ConsumerFactory<K, V> consumerFactory;
private final List<KafkaMessageListenerContainer<K, V>> containers = new ArrayList<>();
private int concurrency = 1;
@@ -70,9 +68,8 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
*/
public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
ContainerProperties containerProperties) {
super(containerProperties);
super(consumerFactory, containerProperties);
Assert.notNull(consumerFactory, "A ConsumerFactory must be provided");
this.consumerFactory = consumerFactory;
}
public int getConcurrency() {

5
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -102,8 +102,6 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
private final AbstractMessageListenerContainer<K, V> container;
private final ConsumerFactory<K, V> consumerFactory;
private final TopicPartitionInitialOffset[] topicPartitions;
private volatile ListenerConsumer listenerConsumer;
@@ -159,10 +157,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
KafkaMessageListenerContainer(AbstractMessageListenerContainer<K, V> container,
ConsumerFactory<K, V> 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);
}

4
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -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<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> 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<ConsumerRecords<Integer, String>>() {
@@ -407,6 +406,7 @@ public class ConcurrentMessageListenerContainerTests {
});
ContainerProperties containerProps = new ContainerProperties(topic1PartitionS);
containerProps.setGroupId("grp");
containerProps.setMessageListener((MessageListener<Integer, String>) message -> { });
ConcurrentMessageListenerContainer<Integer, String> container =

18
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -499,7 +499,7 @@ public class KafkaMessageListenerContainerTests {
public void testRecordAckMock() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> 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<TopicPartition, List<ConsumerRecord<Integer, String>>> 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<Integer, String> messageListener = spy(
@@ -565,7 +566,7 @@ public class KafkaMessageListenerContainerTests {
private void testRecordAckMockForeignThreadGuts(AckMode ackMode) throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> 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<TopicPartition, List<ConsumerRecord<Integer, String>>> 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<Acknowledgment> acks = new ArrayList<>();
@@ -627,7 +629,7 @@ public class KafkaMessageListenerContainerTests {
public void testNonResponsiveConsumerEvent() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(isNull(), eq(""), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq(""), isNull())).willReturn(consumer);
final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> 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<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> 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<TopicPartition, List<ConsumerRecord<Integer, String>>> 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<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
ConsumerRecords<Integer, String> 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<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> 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<TopicPartition, List<ConsumerRecord<Integer, String>>> 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");

142
spring-kafka/src/test/java/org/springframework/kafka/listener/MissingGroupIdTests.java Normal file
View File

@@ -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<String, String> cf() {
return new DefaultKafkaConsumerFactory<>(
Collections.singletonMap(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaEmbedded.getBrokersAsString()),
new StringDeserializer(), new StringDeserializer());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> 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<String, String> cf() {
return new DefaultKafkaConsumerFactory<>(
Collections.singletonMap(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaEmbedded.getBrokersAsString()),
new StringDeserializer(), new StringDeserializer());
}
@Bean
public KafkaMessageListenerContainer<String, String> container() {
ContainerProperties props = new ContainerProperties("missing.group");
return new KafkaMessageListenerContainer<>(cf(), props);
}
}
@Configuration
@EnableKafka
public static class Config3 {
@Bean
public ConsumerFactory<String, String> cf() {
return new DefaultKafkaConsumerFactory<>(
Collections.singletonMap(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaEmbedded.getBrokersAsString()),
new StringDeserializer(), new StringDeserializer());
}
@Bean
public KafkaMessageListenerContainer<String, String> container() {
ContainerProperties props = new ContainerProperties(new TopicPartitionInitialOffset("missing.group", 0));
props.setMessageListener((MessageListener<String, String>) r -> { });
return new KafkaMessageListenerContainer<>(cf(), props);
}
}
}

4
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java
View File

@@ -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;
}

4
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java
View File

@@ -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;
}