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

GH-189: MessageSource - remove seek after poll

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-kafka/issues/189

If the DKCFactory is being used, force the `max.poll.records` property to 1.
Otherwise, throw an `IllegalArgumentException`.
This commit is contained in:
Gary Russell
2018-01-22 15:39:41 -05:00
committed by Artem Bilan
parent 23bc29958c
commit d738447005
2 changed files with 70 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -47,6 +47,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder
import org.springframework.integration.support.AcknowledgmentCallback;
import org.springframework.integration.support.AcknowledgmentCallbackFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.KafkaMessageHeaders;
@@ -115,17 +116,9 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
Assert.notNull(consumerFactory, "'consumerFactory' must not be null");
Assert.notNull(ackCallbackFactory, "'ackCallbackFactory' must not be null");
this.consumerFactory = consumerFactory;
this.consumerFactory = fixOrRejectConsumerFactory(consumerFactory);
this.ackCallbackFactory = ackCallbackFactory;
this.topics = topics;
Object maxPoll = consumerFactory.getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
if (maxPoll == null || (maxPoll instanceof Number && ((Number) maxPoll).intValue() != 1)
|| (maxPoll instanceof String && Integer.parseInt((String) maxPoll) != 1)) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("It is advisable to set " + ConsumerConfig.MAX_POLL_RECORDS_CONFIG
+ " to 1 to avoid having to seek after each record");
}
}
}
protected String getGroupId() {
@@ -221,6 +214,36 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
this.rawMessageHeader = rawMessageHeader;
}
private ConsumerFactory<K, V> fixOrRejectConsumerFactory(ConsumerFactory<K, V> suppliedConsumerFactory) {
Object maxPoll = suppliedConsumerFactory.getConfigurationProperties()
.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
if (maxPoll == null || (maxPoll instanceof Number && ((Number) maxPoll).intValue() != 1)
|| (maxPoll instanceof String && Integer.parseInt((String) maxPoll) != 1)) {
if (!suppliedConsumerFactory.getClass().getName().equals(DefaultKafkaConsumerFactory.class.getName())) {
throw new IllegalArgumentException("Custom consumer factory is not configured with '"
+ ConsumerConfig.MAX_POLL_RECORDS_CONFIG + " = 1'");
}
if (this.logger.isWarnEnabled()) {
this.logger.warn("'" + ConsumerConfig.MAX_POLL_RECORDS_CONFIG
+ "' has been forced to from " + (maxPoll == null ? "unspecified" : maxPoll)
+ "to 1, to avoid having to seek after each record");
}
Map<String, Object> configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties());
configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
DefaultKafkaConsumerFactory<K, V> fixedConsumerFactory = new DefaultKafkaConsumerFactory<>(configs);
if (suppliedConsumerFactory.getKeyDeserializer() != null) {
fixedConsumerFactory.setKeyDeserializer(suppliedConsumerFactory.getKeyDeserializer());
}
if (suppliedConsumerFactory.getValueDeserializer() != null) {
fixedConsumerFactory.setValueDeserializer(suppliedConsumerFactory.getValueDeserializer());
}
return fixedConsumerFactory;
}
else {
return suppliedConsumerFactory;
}
}
@Override
protected synchronized Object doReceive() {
if (this.consumer == null) {
@@ -240,9 +263,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
}
record = records.iterator().next();
topicPartition = new TopicPartition(record.topic(), record.partition());
if (records.count() > 1) {
this.consumer.seek(topicPartition, record.offset() + 1);
}
}
KafkaAckInfo<K, V> ackInfo = new KafkaAckInfoImpl(record, topicPartition);
AcknowledgmentCallback ackCallback = this.ackCallbackFactory.createCallback(ackInfo);

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

@@ -17,15 +17,16 @@
package org.springframework.integration.kafka.inbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willDoNothing;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -42,6 +43,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -56,7 +58,9 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.support.AcknowledgmentCallback;
import org.springframework.integration.support.AcknowledgmentCallback.Status;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
@@ -87,16 +91,13 @@ public class MessageSourceTests {
willAnswer(i -> paused.get()).given(consumer).paused();
Map<TopicPartition, List<ConsumerRecord>> records1 = new LinkedHashMap<>();
records1.put(topicPartition, Arrays.asList(
new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"),
new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar")));
new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
Map<TopicPartition, List<ConsumerRecord>> records2 = new LinkedHashMap<>();
records2.put(topicPartition, Arrays.asList(
new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"),
new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz")));
new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar")));
Map<TopicPartition, List<ConsumerRecord>> records3 = new LinkedHashMap<>();
records3.put(topicPartition, Arrays.asList(
new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"),
new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux")));
new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz")));
Map<TopicPartition, List<ConsumerRecord>> records4 = new LinkedHashMap<>();
records4.put(topicPartition, Collections.singletonList(
new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux")));
@@ -107,6 +108,8 @@ public class MessageSourceTests {
ConsumerRecords cr5 = new ConsumerRecords(Collections.emptyMap());
given(consumer.poll(anyLong())).willReturn(cr1, cr2, cr3, cr4, cr5);
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);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo");
source.setRawMessageHeader(true);
@@ -136,19 +139,16 @@ public class MessageSourceTests {
inOrder.verify(consumer).paused();
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection());
inOrder.verify(consumer).seek(any(TopicPartition.class), eq(1L));
inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L)));
inOrder.verify(consumer).paused();
inOrder.verify(consumer).resume(anyCollection());
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection());
inOrder.verify(consumer).seek(any(TopicPartition.class), eq(2L));
inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L)));
inOrder.verify(consumer).paused();
inOrder.verify(consumer).resume(anyCollection());
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection());
inOrder.verify(consumer).seek(any(TopicPartition.class), eq(3L));
inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(3L)));
inOrder.verify(consumer).paused();
inOrder.verify(consumer).resume(anyCollection());
@@ -206,6 +206,8 @@ public class MessageSourceTests {
ConsumerRecords cr7 = new ConsumerRecords(Collections.emptyMap());
given(consumer.poll(anyLong())).willReturn(cr1, cr2, cr3, cr4, cr5, cr6, cr7);
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);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo");
@@ -274,8 +276,7 @@ public class MessageSourceTests {
willAnswer(i -> paused.get()).given(consumer).paused();
Map<TopicPartition, List<ConsumerRecord>> records1 = new LinkedHashMap<>();
records1.put(topicPartition, Arrays.asList(
new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"),
new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar")));
new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
ConsumerRecords cr1 = new ConsumerRecords(records1);
Map<TopicPartition, List<ConsumerRecord>> records2 = new LinkedHashMap<>();
records2.put(topicPartition, Collections.singletonList(
@@ -284,6 +285,8 @@ public class MessageSourceTests {
ConsumerRecords cr3 = new ConsumerRecords(Collections.emptyMap());
given(consumer.poll(anyLong())).willReturn(cr1, cr1, cr2, cr2, cr3);
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);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo");
@@ -310,13 +313,11 @@ public class MessageSourceTests {
inOrder.verify(consumer).paused();
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection());
inOrder.verify(consumer).seek(topicPartition, 1L); // returned 2 - seek after poll
inOrder.verify(consumer).seek(topicPartition, 0L); // rollback
inOrder.verify(consumer).paused();
inOrder.verify(consumer).resume(anyCollection());
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection());
inOrder.verify(consumer).seek(topicPartition, 1L); // seek after poll
inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L)));
inOrder.verify(consumer).paused();
inOrder.verify(consumer).resume(anyCollection());
@@ -363,6 +364,8 @@ public class MessageSourceTests {
ConsumerRecords cr3 = new ConsumerRecords(Collections.emptyMap());
given(consumer.poll(anyLong())).willReturn(cr1, cr2, cr1, cr2, cr3);
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);
KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo");
@@ -397,7 +400,6 @@ public class MessageSourceTests {
inOrder.verify(consumer).paused();
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection());
inOrder.verify(consumer).seek(topicPartition, 1L); // returned 2 - seek after poll
inOrder.verify(consumer).paused();
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection()); // in flight
@@ -418,7 +420,6 @@ public class MessageSourceTests {
inOrder.verify(consumer).resume(anyCollection());
inOrder.verify(consumer).poll(anyLong());
inOrder.verify(consumer).pause(anyCollection());
inOrder.verify(consumer).seek(topicPartition, 1L); // seek after poll
inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L)));
inOrder.verify(consumer).paused();
inOrder.verify(consumer).resume(anyCollection());
@@ -433,4 +434,25 @@ public class MessageSourceTests {
inOrder.verifyNoMoreInteractions();
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testMaxPollRecords() {
KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap()));
assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class)
.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1);
source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(
Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)));
assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class)
.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1);
try {
new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) {
}));
fail("Expected exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
}
}
}