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

KMSource: Consumer Property Overrides

Browse Source 
- support property overrides in the message source
This commit is contained in:
Gary Russell
2019-09-18 13:22:11 -04:00
committed by Artem Bilan
parent f6d858c7aa
commit adf91edf05
3 changed files with 15 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
View File

@@ -535,7 +535,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
protected void createConsumer() {
synchronized (this.consumerMonitor) {
this.consumer = this.consumerFactory.createConsumer(this.consumerProperties.getGroupId(),
this.consumerProperties.getClientId(), null);
this.consumerProperties.getClientId(), null, this.consumerProperties.getKafkaConsumerProperties());
ConsumerRebalanceListener providedRebalanceListener = this.consumerProperties
.getConsumerRebalanceListener();
boolean isConsumerAware = providedRebalanceListener instanceof ConsumerAwareRebalanceListener;

4
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java
View File

@@ -60,8 +60,11 @@ public class MessageSourceIntegrationTests {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testSource", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 42);
DefaultKafkaConsumerFactory<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
ConsumerProperties consumerProperties = new ConsumerProperties(TOPIC1);
consumerProperties.getKafkaConsumerProperties()
.setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "2");
final CountDownLatch assigned = new CountDownLatch(1);
consumerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
@@ -103,6 +106,7 @@ public class MessageSourceIntegrationTests {
assertThat(received.getPayload()).isEqualTo("qux");
received = source.receive();
assertThat(received).isNull();
assertThat(KafkaTestUtils.getPropertyValue(source, "consumer.fetcher.minBytes")).isEqualTo(2);
source.destroy();
producerFactory.destroy();
}

20
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java
View File

@@ -116,7 +116,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
ConsumerProperties consumerProperties = new ConsumerProperties("foo");
AtomicBoolean partitionsAssignedCalled = new AtomicBoolean();
AtomicReference<Consumer> partitionsAssignedConsumer = new AtomicReference<>();
@@ -165,7 +165,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
ConsumerProperties consumerProperties = new ConsumerProperties("foo");
AtomicBoolean partitionsAssignedCalled = new AtomicBoolean();
AtomicBoolean partitionsRevokedCalled = new AtomicBoolean();
@@ -248,7 +248,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
ConsumerProperties consumerProperties = new ConsumerProperties("foo");
AtomicInteger callbackCount = new AtomicInteger();
OffsetCommitCallback commitCallback = (offsets, ex) -> {
@@ -380,7 +380,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo"));
Message<?> received1 = source.receive();
@@ -454,7 +454,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
ConsumerProperties consumerProperties = new ConsumerProperties("foo");
consumerProperties.setSyncCommitTimeout(Duration.ofSeconds(30));
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, consumerProperties);
@@ -523,7 +523,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo"));
Message<?> received1 = source.receive();
@@ -627,7 +627,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo"));
source.setRawMessageHeader(true);
@@ -689,7 +689,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 4)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo"), true);
source.setRawMessageHeader(true);
@@ -746,7 +746,7 @@ public class MessageSourceTests {
ConsumerFactory<String, String> consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
KafkaMessageSource<String, String> source = new KafkaMessageSource<>(consumerFactory, new ConsumerProperties(Pattern.compile("[a-zA-Z0-9_]*?foo")));
source.setRawMessageHeader(true);
source.start();
@@ -816,7 +816,7 @@ public class MessageSourceTests {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
.getConfigurationProperties();
given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
TopicPartitionOffset beginningTpo = new TopicPartitionOffset(beginning, null, TopicPartitionOffset.SeekPosition.BEGINNING);
TopicPartitionOffset endTpo = new TopicPartitionOffset(end, null, TopicPartitionOffset.SeekPosition.END);