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

polishing

Browse Source
This commit is contained in:
Mark Fisher
2016-02-03 19:51:17 -05:00
parent 8271478e27
commit 252aee4079
4 changed files with 13 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -153,9 +153,9 @@ 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 boolean DEFAULT_RESET_OFFSETS = false;
private static final StartOffset DEFAULT_START = StartOffset.latest;
private static final StartOffset DEFAULT_START_OFFSET = StartOffset.latest;
private RetryOperations retryOperations;
@@ -278,9 +278,9 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
private Mode mode = Mode.embeddedHeaders;
private boolean resetOffsets = false;
private boolean resetOffsets = DEFAULT_RESET_OFFSETS;
private StartOffset startOffset = StartOffset.latest;
private StartOffset startOffset = DEFAULT_START_OFFSET;
public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress,
String... headersToMap) {
@@ -460,11 +460,11 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
@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 resetOffsets at the latest time, which allows them to receive only messages sent after
// they've been bound
// If the caller provides a consumer group, use it; otherwise an anonymous consumer group
// is generated each time, such that each anonymous binding will receive all messages.
// Consumers reset offsets at the latest time by default, which allows them to receive only
// messages sent after they've been bound. That behavior can be changed with the
// "resetOffsets" and "startOffset" properties.
String consumerGroup = group == null ? "anonymous." + UUID.randomUUID().toString() : group;
long referencePoint = this.startOffset != null ?
startOffset.getReferencePoint() : OffsetRequest.LatestTime();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,7 +31,7 @@ import org.springframework.context.annotation.PropertySource;
*/
@Configuration
@ConditionalOnMissingBean(Binder.class)
@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class,KafkaMessageChannelBinderConfiguration.class})
@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaMessageChannelBinderConfiguration.class})
@ImportAutoConfiguration(KafkaMessageChannelBinderConfiguration.class)
public class KafkaServiceAutoConfiguration {

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

@@ -172,7 +172,6 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
Arrays.fill(ratherBigPayload, (byte) 65);
KafkaTestBinder binder = (KafkaTestBinder) getBinder();
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Properties producerProperties = new Properties();
@@ -352,7 +351,6 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
}
@Test
@SuppressWarnings("unchecked")
public void testEarliest() throws Exception {
@@ -427,13 +425,9 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
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 {
@@ -471,6 +465,6 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
assertThat(receivedMessage3, not(nullValue()));
assertThat(new String(receivedMessage3.getPayload()), equalTo(testPayload3));
binder.unbind(consumerBinding);
binder.unbind(producerBinding);
}
}

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-kafka</artifactId>
<artifactId>spring-cloud-stream-binder-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>