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

requestedAcks propoerty changed to string to allow setting it to "all"

Browse Source 
polishing
This commit is contained in:
Rafal Zukowski
2018-02-14 21:17:27 +01:00
committed by Soby Chacko
parent a654382e99
commit 0a59b4c628
2 changed files with 9 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
View File

@@ -36,6 +36,7 @@ import org.springframework.util.StringUtils;
* @author Marius Bogoevici
* @author Soby Chacko
* @author Gary Russell
* @author Rafal Zukowski
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder")
public class KafkaBinderConfigurationProperties {
@@ -83,7 +84,7 @@ public class KafkaBinderConfigurationProperties {
*/
private int zkConnectionTimeout = 10000;
private int requiredAcks = 1;
private String requiredAcks = "1";
private int replicationFactor = 1;
@@ -219,11 +220,15 @@ public class KafkaBinderConfigurationProperties {
this.maxWait = maxWait;
}
public int getRequiredAcks() {
public String getRequiredAcks() {
return this.requiredAcks;
}
public void setRequiredAcks(int requiredAcks) {
this.requiredAcks = String.valueOf(requiredAcks);
}
public void setRequiredAcks(String requiredAcks) {
this.requiredAcks = requiredAcks;
}

3
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams;
/**
* @author Soby Chacko
* @author Rafal Zukowski
*/
import java.util.HashMap;
import java.util.Map;
@@ -103,7 +104,7 @@ class KafkaStreamsDlqDispatch {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(configurationProperties.getRequiredAcks()));
props.put(ProducerConfig.ACKS_CONFIG, configurationProperties.getRequiredAcks());
if (!ObjectUtils.isEmpty(configurationProperties.getProducerConfiguration())) {
props.putAll(configurationProperties.getProducerConfiguration());
}