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

GH-541 Make reconnect time for Kafka/Rabbit configurable

Browse Source
This commit is contained in:
Marius Bogoevici
2016-05-31 17:19:54 -04:00
committed by Gary Russell
parent aef08a28a3
commit 99db7dff9b
6 changed files with 32 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml
View File

@@ -16,7 +16,7 @@
<properties>
<kafka.version>0.8.2.2</kafka.version>
<curator.version>2.6.0</curator.version>
<spring-integration-kafka.version>1.3.0.RELEASE</spring-integration-kafka.version>
<spring-integration-kafka.version>1.3.1.BUILD-SNAPSHOT</spring-integration-kafka.version>
<rxjava-math.version>1.0.0</rxjava-math.version>
</properties>

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

@@ -31,6 +31,8 @@ public class KafkaConsumerProperties {
private boolean enableDlq;
private int recoveryInterval = 5000;
public boolean isAutoCommitOffset() {
return autoCommitOffset;
}
@@ -70,4 +72,12 @@ public class KafkaConsumerProperties {
public void setAutoCommitOnError(Boolean autoCommitOnError) {
this.autoCommitOnError = autoCommitOnError;
}
public int getRecoveryInterval() {
return recoveryInterval;
}
public void setRecoveryInterval(int recoveryInterval) {
this.recoveryInterval = recoveryInterval;
}
}

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

@@ -484,6 +484,7 @@ public class KafkaMessageChannelBinder extends
? properties.getExtension().getAutoCommitOnError()
: properties.getExtension().isAutoCommitOffset() && properties.getExtension().isEnableDlq();
messageListenerContainer.setAutoCommitOnError(autoCommitOnError);
messageListenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
int concurrency = Math.min(properties.getConcurrency(), listenedPartitions.size());
messageListenerContainer.setConcurrency(concurrency);

10
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java
View File

@@ -50,6 +50,8 @@ public class RabbitConsumerProperties {
private String[] replyHeaderPatterns = new String[] {"STANDARD_REPLY_HEADERS", "*"};
private long recoveryInterval = 5000;
public String getPrefix() {
return prefix;
}
@@ -149,4 +151,12 @@ public class RabbitConsumerProperties {
public void setReplyHeaderPatterns(String[] replyHeaderPatterns) {
this.replyHeaderPatterns = replyHeaderPatterns;
}
public long getRecoveryInterval() {
return recoveryInterval;
}
public void setRecoveryInterval(long recoveryInterval) {
this.recoveryInterval = recoveryInterval;
}
}

5
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -280,7 +280,6 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
listenerContainer.setChannelTransacted(properties.getExtension().isTransacted());
listenerContainer.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
int concurrency = properties.getConcurrency();
concurrency = concurrency > 0 ? concurrency : 1;
listenerContainer.setConcurrentConsumers(concurrency);
@@ -288,8 +287,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
if (maxConcurrency > concurrency) {
listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
}
listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTxSize(properties.getExtension().getTxSize());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(queue.getName() + "-"));
listenerContainer.setQueues(queue);
@@ -545,7 +544,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
@Override
public void doManualAck(LinkedList<MessageHeaders> messageHeadersList) {
Iterator<MessageHeaders> iterator = messageHeadersList.iterator();
Map<Object, Long> channelsToAck = new HashMap<Object, Long>();
Map<Object, Long> channelsToAck = new HashMap<>();
while (iterator.hasNext()) {
MessageHeaders messageHeaders = iterator.next();
if (messageHeaders.containsKey(AmqpHeaders.CHANNEL)) {

8
spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc
View File

@@ -990,6 +990,10 @@ prefix::
A prefix to be added to the name of the `destination` and queues.
+
Default: "".
recoveryInterval::
The interval between connection recovery attempts, in milliseconds.
+
Default: `5000`.
requeueRejected::
Whether delivery failures should be requeued.
+
@@ -1143,6 +1147,10 @@ If set to `true`, it will always auto-commit (if auto-commit is enabled).
If not set (default), it effectively has the same value as `enableDlq`, auto-committing erroneous messages if they are sent to a DLQ, and not committing them otherwise.
+
Default: not set.
recoveryInterval::
The interval between connection recovery attempts, in milliseconds.
+
Default: `5000`.
resetOffsets::
Whether to reset offsets on the consumer to the value provided by `startOffset`.
+