SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-67: Workaround for SK GH-599

Browse Source 
Fixes #67

Spring Kafka currently doesn't support `TPIO.SeekPosition` for initial offsets.
Instead, use 0 and `Long.MAX_VALUE` for `BEGINNING` and `END` respectively.

Resolves #331
This commit is contained in:
Gary Russell
2018-03-06 09:51:27 -05:00
committed by Oleg Zhurakousky
parent bcf15ed3be
commit 8035e25359
2 changed files with 34 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -30,6 +30,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -90,7 +91,6 @@ import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.messaging.MessageChannel;
@@ -358,8 +358,7 @@ public class KafkaMessageChannelBinder extends
}
containerProperties.setIdleEventInterval(extendedConsumerProperties.getExtension().getIdleEventInterval());
int concurrency = Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size());
resetOffsets(extendedConsumerProperties, consumerFactory, groupManagement, topicPartitionInitialOffsets,
containerProperties);
resetOffsets(extendedConsumerProperties, consumerFactory, groupManagement, containerProperties);
@SuppressWarnings("rawtypes")
final ConcurrentMessageListenerContainer<?, ?> messageListenerContainer =
new ConcurrentMessageListenerContainer(consumerFactory, containerProperties) {
@@ -404,11 +403,12 @@ public class KafkaMessageChannelBinder extends
}
/*
* Reset the offsets if needed.
* Reset the offsets if needed; may update the offsets in in the container's
* topicPartitionInitialOffsets.
*/
private void resetOffsets(final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
private void resetOffsets(
final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
final ConsumerFactory<?, ?> consumerFactory, boolean groupManagement,
final TopicPartitionInitialOffset[] topicPartitionInitialOffsets,
final ContainerProperties containerProperties) {
boolean resetOffsets = extendedConsumerProperties.getExtension().isResetOffsets();
@@ -446,8 +446,11 @@ public class KafkaMessageChannelBinder extends
});
}
else if (resetOffsets) {
Arrays.stream(topicPartitionInitialOffsets).map(tpio -> new TopicPartitionInitialOffset(tpio.topic(), tpio.partition(),
"earliest".equals(resetTo) ? SeekPosition.BEGINNING : SeekPosition.END));
Arrays.stream(containerProperties.getTopicPartitions())
.map(tpio -> new TopicPartitionInitialOffset(tpio.topic(), tpio.partition(),
// SK GH-599 "earliest".equals(resetTo) ? SeekPosition.BEGINNING : SeekPosition.END))
"earliest".equals(resetTo) ? 0L : Long.MAX_VALUE))
.collect(Collectors.toList()).toArray(containerProperties.getTopicPartitions());
}
}

35
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java
View File

@@ -18,8 +18,8 @@ package org.springframework.cloud.stream.binder.kafka;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -31,7 +31,6 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
@@ -113,19 +112,17 @@ public class KafkaBinderUnitTests {
}
@Test
@Ignore // SK GH-599
public void testOffsetResetWithManualAssignmentEarliest() throws Exception {
testOffsetResetWithGroupManagement(true, false);
}
@Test
@Ignore // SK GH-599
public void testOffsetResetWithGroupManualAssignmentLatest() throws Throwable {
testOffsetResetWithGroupManagement(false, false);
}
private void testOffsetResetWithGroupManagement(final boolean earliest, boolean groupManage) throws Exception {
final Collection<TopicPartition> partitions = new ArrayList<>();
final List<TopicPartition> partitions = new ArrayList<>();
partitions.add(new TopicPartition("foo", 0));
partitions.add(new TopicPartition("foo", 1));
KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties();
@@ -141,7 +138,7 @@ public class KafkaBinderUnitTests {
}).given(provisioningProvider).getPartitionsForTopic(anyInt(), anyBoolean(), any());
@SuppressWarnings("unchecked")
final Consumer<byte[], byte[]> consumer = mock(Consumer.class);
final CountDownLatch latch = new CountDownLatch(1);
final CountDownLatch latch = new CountDownLatch(2);
willAnswer(i -> {
try {
Thread.sleep(100);
@@ -149,18 +146,20 @@ public class KafkaBinderUnitTests {
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
if (!groupManage) {
latch.countDown();
}
return new ConsumerRecords<>(Collections.emptyMap());
}).given(consumer).poll(anyLong());
willAnswer(i -> {
((org.apache.kafka.clients.consumer.ConsumerRebalanceListener) i.getArgument(1))
.onPartitionsAssigned(partitions);
latch.countDown();
latch.countDown();
return null;
}).given(consumer).subscribe(eq(Collections.singletonList("foo")),
any(org.apache.kafka.clients.consumer.ConsumerRebalanceListener.class));
willAnswer(i -> {
latch.countDown();
return null;
}).given(consumer).seek(any(), anyLong());
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties, provisioningProvider) {
@Override
@@ -211,11 +210,23 @@ public class KafkaBinderUnitTests {
consumerProperties.setInstanceCount(1);
binder.bindConsumer("foo", "bar", channel, consumerProperties);
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
if (earliest) {
verify(consumer).seekToBeginning(partitions);
if (groupManage) {
if (earliest) {
verify(consumer).seekToBeginning(partitions);
}
else {
verify(consumer).seekToEnd(partitions);
}
}
else {
verify(consumer).seekToEnd(partitions);
if (earliest) {
verify(consumer).seek(partitions.get(0), 0L);
verify(consumer).seek(partitions.get(1), 0L);
}
else {
verify(consumer).seek(partitions.get(0), Long.MAX_VALUE);
verify(consumer).seek(partitions.get(1), Long.MAX_VALUE);
}
}
}