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

Binding properties should override boot properties

Browse Source 
- This is especially applicable to `group.id` and `auto.offset.reset` which get set when configuring the consumer properties
 - Spring Boot Kafka properties are updated as binder configuration properties which get the least precedence
 - Update test

Resolves #122
This commit is contained in:
Ilayaperumal Gopinathan
2017-05-16 17:58:09 +05:30
parent f280edc9ce
commit 5c70e2df43
3 changed files with 7 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -324,8 +324,6 @@ public class KafkaMessageChannelBinder extends
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, anonymous ? "latest" : "earliest");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) {
props.putAll(configurationProperties.getConfiguration());
@@ -336,6 +334,8 @@ public class KafkaMessageChannelBinder extends
if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getConfiguration())) {
props.putAll(consumerProperties.getExtension().getConfiguration());
}
props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, anonymous ? "latest" : "earliest");
return new DefaultKafkaConsumerFactory<>(props);
}

2
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java
View File

@@ -86,6 +86,8 @@ public class KafkaBinderAutoConfigurationPropertiesTest {
Map<String, Object> consumerConfigs = (Map<String, Object>) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory);
assertTrue(consumerConfigs.get("key.deserializer").equals(LongDeserializer.class));
assertTrue(consumerConfigs.get("value.deserializer").equals(LongDeserializer.class));
assertTrue(consumerConfigs.get("group.id").equals("test"));
assertTrue(consumerConfigs.get("auto.offset.reset").equals("latest"));
assertTrue((((List<String>) consumerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers)));
}

3
spring-cloud-stream-binder-kafka/src/test/resources/binder-config-autoconfig.properties
View File

@@ -5,3 +5,6 @@ spring.kafka.consumer.valueDeserializer=org.apache.kafka.common.serialization.Lo
spring.kafka.producer.batchSize=10
spring.kafka.bootstrapServers=10.98.09.199:9092,10.98.09.196:9092
spring.kafka.producer.compressionType=snappy
# Test consumer properties
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.group-id=testEmbeddedKafkaApplication