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

GH-306 Add support for start offset and reset flag

Browse Source 
Resolves #306

Fix configuration property passing

Adjust property names, fix environment merging from parent
This commit is contained in:
Marius Bogoevici
2016-02-03 16:23:13 -05:00
committed by Mark Fisher
parent 8f795074e9
commit 8271478e27
7 changed files with 230 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -153,6 +153,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
private static final boolean DEFAULT_AUTO_COMMIT_ENABLED = true;
private static final boolean DEFAULT_RESET_ON_START = false;
private static final StartOffset DEFAULT_START = StartOffset.latest;
private RetryOperations retryOperations;
/**
@@ -274,6 +278,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
private Mode mode = Mode.embeddedHeaders;
private boolean resetOffsets = false;
private StartOffset startOffset = StartOffset.latest;
public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress,
String... headersToMap) {
this.zookeeperConnect = zookeeperConnect;
@@ -434,15 +442,33 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
this.mode = mode;
}
public boolean isResetOffsets() {
return resetOffsets;
}
public void setResetOffsets(boolean resetOffsets) {
this.resetOffsets = resetOffsets;
}
public StartOffset getStartOffset() {
return startOffset;
}
public void setStartOffset(StartOffset startOffset) {
this.startOffset = startOffset;
}
@Override
protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel, Properties properties) {
// If the caller provides a group, use it; otherwise
// usage of a different consumer group each time achieves pub-sub
// but multiple instances of this binding will each get all messages
// PubSub consumers reset at the latest time, which allows them to receive only messages sent after
// PubSub consumers resetOffsets at the latest time, which allows them to receive only messages sent after
// they've been bound
String consumerGroup = group == null ? "anonymous." + UUID.randomUUID().toString() : group;
return createKafkaConsumer(name, inputChannel, properties, consumerGroup, OffsetRequest.LatestTime());
long referencePoint = this.startOffset != null ?
startOffset.getReferencePoint() : OffsetRequest.LatestTime();
return createKafkaConsumer(name, inputChannel, properties, consumerGroup, referencePoint);
}
@Override
@@ -678,6 +704,9 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
// if we have less target partitions than target concurrency, adjust accordingly
messageListenerContainer.setConcurrency(Math.min(maxConcurrency, listenedPartitions.size()));
OffsetManager offsetManager = createOffsetManager(group, referencePoint);
if (resetOffsets) {
offsetManager.resetOffsets(listenedPartitions);
}
messageListenerContainer.setOffsetManager(offsetManager);
messageListenerContainer.setQueueSize(accessor.getProperty(QUEUE_SIZE, defaultQueueSize));
messageListenerContainer.setMaxFetch(accessor.getProperty(FETCH_SIZE, defaultFetchSize));
@@ -883,4 +912,19 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
embeddedHeaders
}
public enum StartOffset {
earliest(OffsetRequest.EarliestTime()),
latest(OffsetRequest.LatestTime());
private final long referencePoint;
StartOffset(long referencePoint) {
this.referencePoint = referencePoint;
}
public long getReferencePoint() {
return referencePoint;
}
}
}

25
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java
View File

@@ -19,7 +19,6 @@ package org.springframework.cloud.stream.binder.kafka.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
import org.springframework.context.annotation.Bean;
@@ -36,7 +35,6 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
*/
@Configuration
@EnableConfigurationProperties(KafkaBinderConfigurationProperties.class)
@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.kafka")
public class KafkaMessageChannelBinderConfiguration {
@@ -73,6 +71,10 @@ public class KafkaMessageChannelBinderConfiguration {
private int offsetUpdateShutdownTimeout;
private boolean resetOffsets = false;
private KafkaMessageChannelBinder.StartOffset startOffset;
@Autowired
private Codec codec;
@@ -119,6 +121,9 @@ public class KafkaMessageChannelBinderConfiguration {
.getReplicationFactor());
kafkaMessageChannelBinder.setDefaultRequiredAcks(kafkaBinderConfigurationProperties.getRequiredAcks());
kafkaMessageChannelBinder.setResetOffsets(resetOffsets);
kafkaMessageChannelBinder.setStartOffset(startOffset);
return kafkaMessageChannelBinder;
}
@@ -202,6 +207,22 @@ public class KafkaMessageChannelBinderConfiguration {
return toConnectionString(this.brokers, this.defaultBrokerPort);
}
public KafkaMessageChannelBinder.StartOffset getStartOffset() {
return startOffset;
}
public void setStartOffset(KafkaMessageChannelBinder.StartOffset startOffset) {
this.startOffset = startOffset;
}
public boolean isResetOffsets() {
return resetOffsets;
}
public void setResetOffsets(boolean resetOffsets) {
this.resetOffsets = resetOffsets;
}
/**
* Converts an array of host values to a comma-separated String.
*

2
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.config;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.test.ImportAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
@@ -30,6 +31,7 @@ import org.springframework.context.annotation.PropertySource;
*/
@Configuration
@ConditionalOnMissingBean(Binder.class)
@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class,KafkaMessageChannelBinderConfiguration.class})
@ImportAutoConfiguration(KafkaMessageChannelBinderConfiguration.class)
public class KafkaServiceAutoConfiguration {

158
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -16,6 +16,10 @@
package org.springframework.cloud.stream.binder.kafka;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertNotNull;
@@ -29,6 +33,7 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import kafka.api.OffsetRequest;
import org.junit.ClassRule;
import org.junit.Test;
@@ -38,16 +43,17 @@ import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.integration.kafka.listener.MessageListener;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import kafka.api.OffsetRequest;
import org.springframework.messaging.support.GenericMessage;
/**
@@ -319,4 +325,152 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
binder.unbind(consumerBinding);
}
@Test
@SuppressWarnings("unchecked")
public void testDefaultConsumerStartsAtLatest() throws Exception {
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(new ZookeeperConnect(kafkaTestSupport.getZkConnectString()),
kafkaTestSupport.getBrokerAddress(), kafkaTestSupport.getZkConnectString());
GenericApplicationContext context = new GenericApplicationContext();
context.refresh();
binder.setApplicationContext(context);
binder.afterPropertiesSet();
DirectChannel output = new DirectChannel();
Properties properties = new Properties();
QueueChannel input1 = new QueueChannel();
String testTopicName = UUID.randomUUID().toString();
binder.bindProducer(testTopicName,output,properties);
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Message<byte[]> receivedMessage1 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage1, is(nullValue()));
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
}
@Test
@SuppressWarnings("unchecked")
public void testEarliest() throws Exception {
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(new ZookeeperConnect(kafkaTestSupport.getZkConnectString()),
kafkaTestSupport.getBrokerAddress(), kafkaTestSupport.getZkConnectString());
GenericApplicationContext context = new GenericApplicationContext();
context.refresh();
binder.setApplicationContext(context);
binder.afterPropertiesSet();
binder.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
DirectChannel output = new DirectChannel();
Properties properties = new Properties();
QueueChannel input1 = new QueueChannel();
String testTopicName = UUID.randomUUID().toString();
binder.bindProducer(testTopicName,output,properties);
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Message<byte[]> receivedMessage1 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage1, not(nullValue()));
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
}
@Test
@SuppressWarnings("unchecked")
public void testReset() throws Exception {
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(new ZookeeperConnect(kafkaTestSupport.getZkConnectString()),
kafkaTestSupport.getBrokerAddress(), kafkaTestSupport.getZkConnectString());
GenericApplicationContext context = new GenericApplicationContext();
context.refresh();
binder.setApplicationContext(context);
binder.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
binder.setResetOffsets(true);
binder.afterPropertiesSet();
DirectChannel output = new DirectChannel();
Properties properties = new Properties();
QueueChannel input1 = new QueueChannel();
String testTopicName = UUID.randomUUID().toString();
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, properties);
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
Binding<MessageChannel> consumerBinding =
binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Message<byte[]> receivedMessage1 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage1, not(nullValue()));
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
binder.unbind(consumerBinding);
String testPayload3 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload3.getBytes()));
consumerBinding =
binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Message<byte[]> receivedMessage4 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage4, not(nullValue()));
assertThat(new String(receivedMessage4.getPayload()), equalTo(testPayload1));
Message<byte[]> receivedMessage5 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage5, not(nullValue()));
assertThat(new String(receivedMessage5.getPayload()), equalTo(testPayload2));
Message<byte[]> receivedMessage6 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage6, not(nullValue()));
assertThat(new String(receivedMessage6.getPayload()), equalTo(testPayload3));
binder.unbind(consumerBinding);
binder.unbind(producerBinding);
}
@Test
@SuppressWarnings("unchecked")
public void testResume() throws Exception {
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(new ZookeeperConnect(kafkaTestSupport.getZkConnectString()),
kafkaTestSupport.getBrokerAddress(), kafkaTestSupport.getZkConnectString());
GenericApplicationContext context = new GenericApplicationContext();
context.refresh();
binder.setApplicationContext(context);
binder.afterPropertiesSet();
binder.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
DirectChannel output = new DirectChannel();
Properties properties = new Properties();
QueueChannel input1 = new QueueChannel();
String testTopicName = UUID.randomUUID().toString();
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, properties);
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Message<byte[]> receivedMessage1 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage1, not(nullValue()));
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
binder.unbind(consumerBinding);
String testPayload3 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload3.getBytes()));
consumerBinding =
binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Message<byte[]> receivedMessage3 = (Message<byte[]>) input1.receive(1000);
assertThat(receivedMessage3, not(nullValue()));
assertThat(new String(receivedMessage3.getPayload()), equalTo(testPayload3));
binder.unbind(consumerBinding);
}
}

1
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java
View File

@@ -51,6 +51,7 @@ import org.springframework.messaging.support.GenericMessage;
*
* @author Gary Russell
* @author Mark Fisher
* @author Marius Bogoevici
*/
abstract public class PartitionCapableBinderTests extends BrokerBinderTests {

2
spring-cloud-stream-samples/source/pom.xml
View File

@@ -24,7 +24,7 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-redis</artifactId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -145,7 +145,7 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
if (useApplicationContextAsParent) {
springApplicationBuilder.parent(context);
}
else if (environment != null && binderConfiguration.isInheritEnvironment()) {
if (useApplicationContextAsParent || (environment != null && binderConfiguration.isInheritEnvironment())) {
StandardEnvironment binderEnvironment = new StandardEnvironment();
binderEnvironment.merge(environment);
springApplicationBuilder.environment(binderEnvironment);