From adf91edf050d6d496192a3c670a75eb15ecff8a0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 18 Sep 2019 13:22:11 -0400 Subject: [PATCH] KMSource: Consumer Property Overrides - support property overrides in the message source --- .../kafka/inbound/KafkaMessageSource.java | 2 +- .../MessageSourceIntegrationTests.java | 4 ++++ .../kafka/inbound/MessageSourceTests.java | 20 +++++++++---------- 3 files changed, 15 insertions(+), 11 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index 6e5fdde330..827ac9fad5 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -535,7 +535,7 @@ public class KafkaMessageSource extends AbstractMessageSource 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; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java index 5459a37196..49cb4f9fc2 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java @@ -60,8 +60,11 @@ public class MessageSourceIntegrationTests { Map 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 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(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index 1b684adeb6..713fbbaf87 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -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 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 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(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);