Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-75: ContainerFactory Improvements

Browse Source 
Resolves #75

Add `syncCommit` and `commitCallback` to container factory.
This commit is contained in:
Gary Russell
2016-05-13 12:04:40 -04:00
parent d93d91b339
commit ead9783087
5 changed files with 90 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
build.gradle
View File

@@ -78,7 +78,7 @@ subprojects { subproject ->
mockitoVersion = '1.9.5'
scalaVersion = '2.11'
springRetryVersion = '1.1.2.RELEASE'
springVersion = '4.2.5.RELEASE'
springVersion = '4.2.6.RELEASE'
idPrefix = 'kafka'
@@ -154,6 +154,7 @@ project ('spring-kafka-test') {
description = 'Spring Kafka Test Support'
dependencies {
compile "org.springframework:spring-beans:$springVersion"
compile "org.springframework:spring-test:$springVersion"
compile "org.springframework.retry:spring-retry:$springRetryVersion"
compile "org.apache.kafka:kafka_$scalaVersion:$kafkaVersion"

45
spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java
View File

@@ -31,7 +31,9 @@ import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.util.Assert;
/**
* Kafka testing utilities.
@@ -129,4 +131,47 @@ public final class KafkaTestUtils {
return received;
}
/**
* Uses nested {@link DirectFieldAccessor}s to obtain a property using dotted notation to traverse fields; e.g.
* "foo.bar.baz" will obtain a reference to the baz field of the bar field of foo. Adopted from Spring Integration.
* @param root The object.
* @param propertyPath The path.
* @return The field.
*/
public static Object getPropertyValue(Object root, String propertyPath) {
Object value = null;
DirectFieldAccessor accessor = new DirectFieldAccessor(root);
String[] tokens = propertyPath.split("\\.");
for (int i = 0; i < tokens.length; i++) {
value = accessor.getPropertyValue(tokens[i]);
if (value != null) {
accessor = new DirectFieldAccessor(value);
}
else if (i == tokens.length - 1) {
return null;
}
else {
throw new IllegalArgumentException("intermediate property '" + tokens[i] + "' is null");
}
}
return value;
}
/**
* A typed version of {@link #getPropertyValue(Object, String)}.
* @param root the object.
* @param propertyPath the path.
* @param type the type to cast the object to
* @return the field value.
* @see #getPropertyValue(Object, String)
*/
@SuppressWarnings("unchecked")
public static <T> T getPropertyValue(Object root, String propertyPath, Class<T> type) {
Object value = getPropertyValue(root, propertyPath);
if (value != null) {
Assert.isAssignable(type, value.getClass());
}
return (T) value;
}
}

29
spring-kafka/src/main/java/org/springframework/kafka/config/SimpleKafkaListenerContainerFactory.java
View File

@@ -19,6 +19,7 @@ package org.springframework.kafka.config;
import java.util.Collection;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
@@ -47,6 +48,10 @@ public class SimpleKafkaListenerContainerFactory<K, V>
private ConsumerRebalanceListener consumerRebalanceListener;
private OffsetCommitCallback commitCallback;
private Boolean syncCommits;
/**
* Specify the container concurrency.
* @param concurrency the number of consumers to create.
@@ -75,6 +80,24 @@ public class SimpleKafkaListenerContainerFactory<K, V>
this.consumerRebalanceListener = consumerRebalanceListener;
}
/**
* Specify the commit callback.
* @param commitCallback the callback to set.
* @see ConcurrentMessageListenerContainer#setCommitCallback(OffsetCommitCallback)
*/
public void setCommitCallback(OffsetCommitCallback commitCallback) {
this.commitCallback = commitCallback;
}
/**
* Specifiy whether or not to use sync commits.
* @param syncCommits the sync commits to set.
* @see ConcurrentMessageListenerContainer#setSyncCommits(boolean)
*/
public void setSyncCommits(Boolean syncCommits) {
this.syncCommits = syncCommits;
}
@Override
protected ConcurrentMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
Collection<TopicPartition> topicPartitions = endpoint.getTopicPartitions();
@@ -107,6 +130,12 @@ public class SimpleKafkaListenerContainerFactory<K, V>
if (this.consumerRebalanceListener != null) {
instance.setConsumerRebalanceListener(this.consumerRebalanceListener);
}
if (this.commitCallback != null) {
instance.setCommitCallback(this.commitCallback);
}
if (this.syncCommits != null) {
instance.setSyncCommits(this.syncCommits);
}
}
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
View File

@@ -64,7 +64,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
private OffsetCommitCallback commitCallback;
private boolean syncCommits;
private boolean syncCommits = true;
/**
* Construct an instance with the supplied configuration properties and specific

19
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.kafka.annotation;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import java.util.Collection;
import java.util.Map;
@@ -26,7 +27,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -139,6 +140,9 @@ public class EnableKafkaIntegrationTests {
this.registry.start();
assertThat(listenerContainer.isRunning()).isTrue();
listenerContainer.stop();
assertThat(KafkaTestUtils.getPropertyValue(listenerContainer, "syncCommits", Boolean.class)).isFalse();
assertThat(KafkaTestUtils.getPropertyValue(listenerContainer, "commitCallback")).isNotNull();
assertThat(KafkaTestUtils.getPropertyValue(listenerContainer, "consumerRebalanceListener")).isNotNull();
}
@Test
@@ -187,7 +191,7 @@ public class EnableKafkaIntegrationTests {
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaListenerContainerFactory() {
kafkaListenerContainerFactory() {
SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
@@ -195,7 +199,7 @@ public class EnableKafkaIntegrationTests {
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaJsonListenerContainerFactory() {
kafkaJsonListenerContainerFactory() {
SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setMessageConverter(new StringJsonMessageConverter());
@@ -204,7 +208,7 @@ public class EnableKafkaIntegrationTests {
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaManualAckListenerContainerFactory() {
kafkaManualAckListenerContainerFactory() {
SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
factory.setConsumerFactory(manualConsumerFactory());
factory.setAckMode(AckMode.MANUAL_IMMEDIATE);
@@ -213,16 +217,19 @@ public class EnableKafkaIntegrationTests {
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaAutoStartFalseListenerContainerFactory() {
kafkaAutoStartFalseListenerContainerFactory() {
SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setAutoStartup(false);
factory.setSyncCommits(false);
factory.setCommitCallback(mock(OffsetCommitCallback.class));
factory.setConsumerRebalanceListener(mock(ConsumerRebalanceListener.class));
return factory;
}
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaRebalanceListenerContainerFactory() {
kafkaRebalanceListenerContainerFactory() {
SimpleKafkaListenerContainerFactory<Integer, String> factory = new SimpleKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConsumerRebalanceListener(consumerRebalanceListener());