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

INTEXT 156-157

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-156
https://jira.spring.io/browse/INTEXT-157

- `KafkaMessageListenerContainer#stop()` returns only after the container has been effectively stopped, i.e. no more messages are dispatched to the listeners;
- add a configurable timeout property to prevent the previous operation from blocking indefinitely;
- create separate XSDs for 1.0 and 1.1 schemas;

Corrections after review

- removed 1.0 schema
- removed time-sensitive tests

Corrections after review

- removed 1.0 schema
- removed time-sensitive tests
- removed unused logger in QueueingMessageListenerInvoker

Clear messages after stopping

Clear partitionsByBrokersMap

Polishing imports and properties order
This commit is contained in:
Marius Bogoevici
2015-03-25 21:26:43 +02:00
committed by Artem Bilan
parent 3e6a796e85
commit 1e6e7b9bbf
10 changed files with 85 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java
View File

@@ -44,6 +44,7 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt
String errorHandler = element.getAttribute("error-handler");
String taskExecutor = element.getAttribute("task-executor");
String concurrency = element.getAttribute("concurrency");
String stopTimeout = element.getAttribute("stop-timeout");
String maxFetch = element.getAttribute("max-fetch");
String queueSize = element.getAttribute("queue-size");
@@ -51,10 +52,11 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt
(StringUtils.hasText(connectionFactory) || StringUtils.hasText(topics)
|| StringUtils.hasText(offsetManager) || StringUtils.hasText(errorHandler)
|| StringUtils.hasText(taskExecutor) || StringUtils.hasText(concurrency)
|| StringUtils.hasText(maxFetch) || StringUtils.hasText(queueSize))) {
|| StringUtils.hasText(maxFetch) || StringUtils.hasText(queueSize)
|| StringUtils.hasText(stopTimeout))) {
parserContext.getReaderContext().error("The 'listener-container' is mutually exclusive with " +
"'connection-factory', 'topics', 'offset-manager', 'error-handler', 'task-executor', " +
"'concurrency', 'max-fetch' and 'queue-size'.", element);
"'concurrency', 'stop-timeout', 'max-fetch' and 'queue-size'.", element);
}
if (StringUtils.hasText(container)) {
@@ -81,6 +83,7 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt
containerBuilder, element, "task-executor", "fetchTaskExecutor");
IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "concurrency");
IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "max-fetch");
IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "stop-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "queue-size");
builder.addConstructorArgValue(containerBuilder.getBeanDefinition());

7
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -19,9 +19,6 @@ package org.springframework.integration.kafka.inbound;
import java.util.HashMap;
import java.util.Map;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.kafka.core.KafkaMessageMetadata;
@@ -37,6 +34,9 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
/**
* @author Marius Bogoevici
*/
@@ -76,7 +76,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp
* adapter inserts a 'kafka_acknowledgment` header allowing the user to manually
* commit the offset using the {@link Acknowledgment#acknowledge()} method.
* Default 'true'.
*
* @param autoCommitOffset false to not auto-commit (default true).
*/
public void setAutoCommitOffset(boolean autoCommitOffset) {

43
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java
View File

@@ -22,17 +22,15 @@ import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import com.gs.collections.api.block.procedure.Procedure;
import com.gs.collections.api.block.procedure.Procedure2;
import com.gs.collections.api.map.MutableMap;
import com.gs.collections.impl.factory.Maps;
import org.springframework.context.Lifecycle;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.Assert;
import com.gs.collections.api.block.procedure.Procedure2;
import com.gs.collections.api.map.MutableMap;
import com.gs.collections.impl.factory.Maps;
/**
* Dispatches {@link KafkaMessage}s to a {@link MessageListener}. Messages may be
* processed concurrently, according to the {@code concurrency} settings, but messages
@@ -40,7 +38,7 @@ import org.springframework.util.Assert;
*
* @author Marius Bogoevici
*/
class ConcurrentMessageListenerDispatcher implements Lifecycle {
class ConcurrentMessageListenerDispatcher {
public static final CustomizableThreadFactory THREAD_FACTORY = new CustomizableThreadFactory("dispatcher-");
@@ -54,18 +52,18 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle {
private final int consumers;
private volatile boolean running;
private final Object delegateListener;
private final ErrorHandler errorHandler;
private final OffsetManager offsetManager;
private MutableMap<Partition, QueueingMessageListenerInvoker> delegates;
private final int queueSize;
private volatile boolean running;
private MutableMap<Partition, QueueingMessageListenerInvoker> delegates;
private Executor taskExecutor;
public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler,
@@ -87,33 +85,26 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle {
this.queueSize = queueSize;
}
@Override
public void start() {
synchronized (lifecycleMonitor) {
if (!isRunning()) {
if (!this.running) {
initializeAndStartDispatching();
this.running = true;
}
}
}
@Override
public void stop() {
public void stop(int stopTimeout) {
synchronized (lifecycleMonitor) {
if (isRunning()) {
if (this.running) {
this.running = false;
delegates.flip().keyBag().toSet().forEach(stopDelegateProcedure);
delegates.flip().keyBag().toSet().forEachWith(stopDelegateProcedure, stopTimeout);
}
}
}
@Override
public boolean isRunning() {
return running;
}
public void dispatch(KafkaMessage message) {
if (isRunning()) {
if (this.running) {
delegates.get(message.getMetadata().getPartition()).enqueue(message);
}
}
@@ -141,11 +132,11 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle {
}
@SuppressWarnings("serial")
private static class StopDelegateProcedure implements Procedure<QueueingMessageListenerInvoker> {
private static class StopDelegateProcedure implements Procedure2<QueueingMessageListenerInvoker, Integer> {
@Override
public void value(QueueingMessageListenerInvoker delegate) {
delegate.stop();
public void value(QueueingMessageListenerInvoker delegate, Integer stopTimeout) {
delegate.stop(stopTimeout);
}
}

20
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -71,6 +71,8 @@ import kafka.common.ErrorMapping;
*/
public class KafkaMessageListenerContainer implements SmartLifecycle {
private static final int DEFAULT_STOP_TIMEOUT = 1000;
private static final Log log = LogFactory.getLog(KafkaMessageListenerContainer.class);
public static final Function<Map.Entry<Partition, ?>, Partition> keyFunction = Functions.getKeyFunction();
@@ -105,6 +107,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
private int queueSize = 1024;
private int stopTimeout = DEFAULT_STOP_TIMEOUT;
private Object messageListener;
private ErrorHandler errorHandler = new LoggingErrorHandler();
@@ -176,6 +180,19 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
this.concurrency = concurrency;
}
/**
* The timeout for waiting for each concurrent {@link MessageListener} to finish on stopping.
* @param stopTimeout timeout in milliseconds
* @since 1.1
*/
public void setStopTimeout(int stopTimeout) {
this.stopTimeout = stopTimeout;
}
public int getStopTimeout() {
return stopTimeout;
}
public Executor getFetchTaskExecutor() {
return fetchTaskExecutor;
}
@@ -243,7 +260,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
catch (IOException e) {
log.error("Error while flushing:", e);
}
this.messageDispatcher.stop();
this.messageDispatcher.stop(stopTimeout);
}
}
if (callback != null) {
@@ -268,6 +285,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
this.messageDispatcher = new ConcurrentMessageListenerDispatcher(messageListener, errorHandler,
Arrays.asList(partitions), offsetManager, concurrency, queueSize);
this.messageDispatcher.start();
partitionsByBrokerMap.clear();
partitionsByBrokerMap.putAll(partitionsAsList.groupBy(getLeader));
if (fetchTaskExecutor == null) {
fetchTaskExecutor = Executors.newFixedThreadPool(partitionsByBrokerMap.size());

37
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java
View File

@@ -18,8 +18,9 @@ package org.springframework.integration.kafka.listener;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.springframework.context.Lifecycle;
import org.springframework.integration.kafka.core.KafkaMessage;
/**
@@ -28,11 +29,7 @@ import org.springframework.integration.kafka.core.KafkaMessage;
*
* @author Marius Bogoevici
*/
class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
private BlockingQueue<KafkaMessage> messages;
private volatile boolean running = false;
class QueueingMessageListenerInvoker implements Runnable {
private final MessageListener messageListener;
@@ -42,6 +39,12 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
private final ErrorHandler errorHandler;
private BlockingQueue<KafkaMessage> messages;
private volatile boolean running = false;
private volatile CountDownLatch shutdownLatch = null;
public QueueingMessageListenerInvoker(int capacity, OffsetManager offsetManager, Object delegate,
ErrorHandler errorHandler) {
if (delegate instanceof MessageListener) {
@@ -90,19 +93,20 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
}
}
@Override
public void start() {
this.running = true;
}
@Override
public void stop() {
public void stop(long stopTimeout) {
shutdownLatch = new CountDownLatch(1);
this.running = false;
}
@Override
public boolean isRunning() {
return this.running;
try {
shutdownLatch.await(stopTimeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
messages.clear();
}
/**
@@ -115,7 +119,7 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
while (this.running) {
try {
KafkaMessage message = messages.take();
if (isRunning()) {
if (this.running) {
try {
if (messageListener != null) {
messageListener.onMessage(message);
@@ -141,6 +145,9 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
wasInterrupted = true;
}
}
if (shutdownLatch != null) {
shutdownLatch.countDown();
}
if (wasInterrupted) {
Thread.currentThread().interrupt();
}

4
spring-integration-kafka/src/main/resources/META-INF/spring.schemas
View File

@@ -1,2 +1,2 @@
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.0.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.1.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd

11
spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd → spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd
View File

@@ -547,7 +547,7 @@
<xsd:documentation>
A 'org.springframework.integration.kafka.listener.KafkaMessageListenerContainer' bean reference.
Mutually exclusive with 'connection-factory', 'topics', 'offset-manager', 'error-handler',
'task-executor', 'concurrency', 'max-fetch', 'queue-size'.
'task-executor', 'concurrency', 'stop-timeout', 'max-fetch', 'queue-size'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -632,6 +632,15 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="stop-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The maximum amount of time (in milliseconds) to wait for each 'org.springframework.integration.kafka.listener.QueueingMessageListenerInvoker' to finish before stopping.
Defaults to '1000'.
Mutually exclusive with 'listener-container'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="max-fetch" type="xsd:string">
<xsd:annotation>
<xsd:documentation>

1
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml
View File

@@ -46,6 +46,7 @@
payload-decoder="payloadDecoder"
offset-manager="offsetManager"
task-executor="executor"
stop-timeout="${stop.timeout:5000}"
queue-size="${queue.size:1024}"
concurrency="${concurrency:10}"
max-fetch="${max.fetch:1000}"

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
View File

@@ -121,7 +121,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
assertEquals(10, container.getConcurrency());
assertEquals(1000, container.getMaxFetch());
assertEquals(1024, container.getQueueSize());
assertEquals(1024, container.getQueueSize());
assertEquals(5000, container.getStopTimeout());
assertArrayEquals(new String[] {"foo", "bar"}, TestUtils.getPropertyValue(container, "topics", String[].class));
assertOverrides(this.kafkaListener, false, false, false, true);
assertOverrides(this.withMBFactoryOverrideAndId, true, true, false, false);

11
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java
View File

@@ -29,6 +29,10 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.gs.collections.api.multimap.list.MutableListMultimap;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
import kafka.message.NoCompressionCodec$;
import org.junit.Rule;
import org.junit.Test;
@@ -44,12 +48,6 @@ import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import com.gs.collections.api.multimap.list.MutableListMultimap;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
import kafka.message.NoCompressionCodec$;
/**
* @author Marius Bogoevici
*/
@@ -220,4 +218,5 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), equalTo(String.valueOf(20)));
}
}
}