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

Add Reactor based ringbuffer

Browse Source 
restart support

blocking stop

Test fixes

Ensure that an executor can be set externally

Polishing according PR comments
This commit is contained in:
Marius Bogoevici
2015-05-21 20:10:04 -04:00
committed by Artem Bilan
parent d8df11c2aa
commit 38856ee984
5 changed files with 124 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,9 +16,11 @@
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;
@@ -31,12 +33,12 @@ import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.MutableMessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.Assert;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
/**
* @author Marius Bogoevici
*/
@@ -168,7 +170,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp
@Override
public void doOnMessage(Object key, Object payload, KafkaMessageMetadata metadata) {
KafkaMessageDrivenChannelAdapter.this.sendMessage(toMessage(key, payload, metadata, null));
sendMessage(toMessage(key, payload, metadata, null));
}
}
@@ -184,95 +186,38 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp
@Override
public void doOnMessage(Object key, Object payload, KafkaMessageMetadata metadata,
Acknowledgment acknowledgment) {
KafkaMessageDrivenChannelAdapter.this.sendMessage(toMessage(key, payload, metadata, acknowledgment));
}
sendMessage(toMessage(key, payload, metadata, acknowledgment));
}
}
private Message<Object> toMessage(Object key, Object payload, KafkaMessageMetadata metadata,
Acknowledgment acknowledgment) {
final Map<String, Object> headers = new HashMap<String, Object>();
final MessageHeaderAccessor headerAccessor = new MessageHeaderAccessor();
headers.put(KafkaHeaders.MESSAGE_KEY, key);
headers.put(KafkaHeaders.TOPIC, metadata.getPartition().getTopic());
headers.put(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId());
headers.put(KafkaHeaders.OFFSET, metadata.getOffset());
headers.put(KafkaHeaders.NEXT_OFFSET, metadata.getNextOffset());
headerAccessor.setHeader(KafkaHeaders.MESSAGE_KEY, key);
headerAccessor.setHeader(KafkaHeaders.TOPIC, metadata.getPartition().getTopic());
headerAccessor.setHeader(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId());
headerAccessor.setHeader(KafkaHeaders.OFFSET, metadata.getOffset());
headerAccessor.setHeader(KafkaHeaders.NEXT_OFFSET, metadata.getNextOffset());
// pre-set the message id header if set to not generate
if (!this.generateMessageId) {
headers.put(MessageHeaders.ID, MessageHeaders.ID_VALUE_NONE);
}
headerAccessor.setLeaveMutable(!(this.generateMessageId || this.generateTimestamp));
// pre-set the timestamp header if set to not generate
if (!this.generateTimestamp) {
headers.put(MessageHeaders.TIMESTAMP, -1L);
}
if (!this.autoCommitOffset) {
headers.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);
headerAccessor.setHeader(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);
}
if (this.useMessageBuilderFactory) {
return getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.copyHeaders(headerAccessor.toMessageHeaders())
.build();
}
else {
return new KafkaMessage(payload, headers);
}
}
/**
* Special subclass of {@link Message}. It is used for lower message generation overhead, unless the default
* strategy of the outer class is set via {@link #setUseMessageBuilderFactory(boolean)}
* @since 1.1
*/
private class KafkaMessage implements Message<Object> {
private final Object payload;
private final MessageHeaders messageHeaders;
public KafkaMessage(Object payload, Map<String, Object> headers) {
this.payload = payload;
this.messageHeaders = new KafkaMessageHeaders(headers, generateMessageId, generateTimestamp);
}
@Override
public Object getPayload() {
return this.payload;
}
@Override
public MessageHeaders getHeaders() {
return this.messageHeaders;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder(getClass().getSimpleName());
sb.append(" [payload=");
if (this.payload instanceof byte[]) {
sb.append("byte[").append(((byte[]) this.payload).length).append("]");
}
else {
sb.append(this.payload);
}
sb.append(", headers=").append(this.messageHeaders).append("]");
return sb.toString();
}
}
@SuppressWarnings("serial")
private class KafkaMessageHeaders extends MessageHeaders {
public KafkaMessageHeaders(Map<String, Object> headers, boolean generateId, boolean generateTimestamp) {
super(headers, generateId ? null : ID_VALUE_NONE, generateTimestamp ? null : -1L);
return MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders());
}
}

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

@@ -22,6 +22,8 @@ import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import com.gs.collections.api.block.procedure.Procedure;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -43,8 +45,6 @@ import com.gs.collections.impl.factory.Maps;
*/
class ConcurrentMessageListenerDispatcher {
public static final CustomizableThreadFactory THREAD_FACTORY = new CustomizableThreadFactory("dispatcher-");
private static final Log log = LogFactory.getLog(ConcurrentMessageListenerDispatcher.class);
private static final StartDelegateProcedure startDelegateProcedure = new StartDelegateProcedure();
@@ -65,14 +65,14 @@ class ConcurrentMessageListenerDispatcher {
private final int queueSize;
private final Executor taskExecutor;
private volatile boolean running;
private MutableMap<Partition, QueueingMessageListenerInvoker> delegates;
private Executor taskExecutor;
public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler,
Collection<Partition> partitions, OffsetManager offsetManager, int consumers, int queueSize, Executor taskExecutor) {
Collection<Partition> partitions, OffsetManager offsetManager, int consumers, int queueSize, Executor taskExecutor) {
Assert.isTrue
(delegateListener instanceof MessageListener
|| delegateListener instanceof AcknowledgingMessageListener,
@@ -120,22 +120,20 @@ class ConcurrentMessageListenerDispatcher {
// allocate delegate instances index them
List<QueueingMessageListenerInvoker> delegateList = new ArrayList<QueueingMessageListenerInvoker>(consumers);
for (int i = 0; i < consumers; i++) {
QueueingMessageListenerInvoker blockingQueueMessageListenerInvoker =
new QueueingMessageListenerInvoker(queueSize, offsetManager, delegateListener, errorHandler);
delegateList.add(blockingQueueMessageListenerInvoker);
QueueingMessageListenerInvoker queueingMessageListenerInvoker =
new QueueingMessageListenerInvoker(queueSize, offsetManager, delegateListener, errorHandler, taskExecutor);
delegateList.add(queueingMessageListenerInvoker);
}
// evenly distribute partitions across delegates
delegates = Maps.mutable.of();
int i = 0;
for (Partition partition : partitions) {
delegates.put(partition, delegateList.get((i++) % consumers));
}
// initialize task executor
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newFixedThreadPool(consumers, THREAD_FACTORY);
}
// start dispatchers
delegates.flip().keyBag().toSet().forEachWith(startDelegateProcedure, taskExecutor);
delegates.flip().keyBag().toSet().forEach(startDelegateProcedure);
}
@SuppressWarnings("serial")
@@ -145,9 +143,10 @@ class ConcurrentMessageListenerDispatcher {
public void value(QueueingMessageListenerInvoker delegate, Integer stopTimeout) {
try {
delegate.stop(stopTimeout);
} catch (Exception e) {
}
catch (Exception e) {
// ignore the exception, but log it
if(log.isInfoEnabled()) {
if (log.isInfoEnabled()) {
log.info("Exception thrown while stopping dispatcher:", e);
}
}
@@ -156,12 +155,11 @@ class ConcurrentMessageListenerDispatcher {
}
@SuppressWarnings("serial")
private static class StartDelegateProcedure implements Procedure2<QueueingMessageListenerInvoker, Executor> {
private static class StartDelegateProcedure implements Procedure<QueueingMessageListenerInvoker> {
@Override
public void value(QueueingMessageListenerInvoker delegate, Executor executor) {
public void value(QueueingMessageListenerInvoker delegate) {
delegate.start();
executor.execute(delegate);
}
}

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

@@ -16,20 +16,29 @@
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.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.core.task.support.ExecutorServiceAdapter;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
import reactor.core.processor.RingBufferProcessor;
/**
* Invokes a delegate {@link MessageListener} for all the messages passed to it, storing them
* in an internal queue.
*
* @author Marius Bogoevici
* @author Stephane Maldini
*/
class QueueingMessageListenerInvoker implements Runnable {
class QueueingMessageListenerInvoker {
private final MessageListener messageListener;
@@ -39,14 +48,19 @@ class QueueingMessageListenerInvoker implements Runnable {
private final ErrorHandler errorHandler;
private BlockingQueue<KafkaMessage> messages;
private final int capacity;
private final ExecutorService executorService;
private RingBufferProcessor<KafkaMessage> ringBufferProcessor;
private volatile boolean running = false;
private volatile CountDownLatch shutdownLatch = null;
private volatile CountDownLatch shutdownLatch;
public QueueingMessageListenerInvoker(int capacity, OffsetManager offsetManager, Object delegate,
ErrorHandler errorHandler) {
public QueueingMessageListenerInvoker(int capacity, final OffsetManager offsetManager, Object delegate,
final ErrorHandler errorHandler, Executor executor) {
this.capacity = capacity;
if (delegate instanceof MessageListener) {
this.messageListener = (MessageListener) delegate;
this.acknowledgingMessageListener = null;
@@ -62,95 +76,92 @@ class QueueingMessageListenerInvoker implements Runnable {
}
this.offsetManager = offsetManager;
this.errorHandler = errorHandler;
this.messages = new ArrayBlockingQueue<KafkaMessage>(capacity);
if (executor != null) {
this.executorService = new ExecutorServiceAdapter(new ConcurrentTaskExecutor(executor));
}
else {
this.executorService = null;
}
}
/**
* Add a message to the queue, blocking if the queue has reached its maximum capacity.
* Interrupts will be ignored for as long as the component's {@code running} flag is set to true, but will
* be deferred for when the method returns.
*
* @param message the KafkaMessage to add
*/
public void enqueue(KafkaMessage message) {
boolean wasInterruptedWhileRunning = false;
if (this.running) {
boolean added = false;
// handle the case when the thread is interrupted while the adapter is still running
// retry adding the message to the queue until either we succeed, or the adapter is stopped
while (!added && this.running) {
try {
this.messages.put(message);
added = true;
}
catch (InterruptedException e) {
// we ignore the interruption signal if we are still running, but pass it on if we are stopped
wasInterruptedWhileRunning = true;
}
}
}
if (wasInterruptedWhileRunning) {
Thread.currentThread().interrupt();
ringBufferProcessor.onNext(message);
}
}
public void start() {
this.running = true;
ExecutorService service = executorService != null ? executorService : Executors.newSingleThreadExecutor();
this.ringBufferProcessor = RingBufferProcessor.share(service, capacity);
this.ringBufferProcessor.subscribe(new KafkaMessageDispatchingSubscriber());
}
public void stop(long stopTimeout) {
shutdownLatch = new CountDownLatch(1);
this.running = false;
try {
shutdownLatch.await(stopTimeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
messages.clear();
}
/**
* Runs uninterruptibly as long as {@code running} is true, but if interrupted, will defer
* propagating the interruption flag at the end.
*/
@Override
public void run() {
boolean wasInterrupted = false;
while (this.running) {
if (ringBufferProcessor != null) {
ringBufferProcessor.onComplete();
ringBufferProcessor = null;
shutdownLatch = new CountDownLatch(1);
try {
KafkaMessage message = messages.take();
if (this.running) {
try {
if (messageListener != null) {
messageListener.onMessage(message);
}
else {
acknowledgingMessageListener.onMessage(message, new DefaultAcknowledgment(offsetManager, message));
}
}
catch (Exception e) {
if (errorHandler != null) {
errorHandler.handle(e, message);
}
}
finally {
if (messageListener != null) {
offsetManager.updateOffset(message.getMetadata().getPartition(),
message.getMetadata().getNextOffset());
}
}
}
shutdownLatch.await(stopTimeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
wasInterrupted = true;
Thread.currentThread().interrupt();
}
}
if (shutdownLatch != null) {
shutdownLatch.countDown();
}
if (wasInterrupted) {
Thread.currentThread().interrupt();
}
}
private class KafkaMessageDispatchingSubscriber implements Subscriber<KafkaMessage> {
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(KafkaMessage kafkaMessage) {
try {
if (messageListener != null) {
messageListener.onMessage(kafkaMessage);
}
else {
acknowledgingMessageListener.onMessage(kafkaMessage, new DefaultAcknowledgment(offsetManager, kafkaMessage));
}
}
catch (Exception e) {
// we handle errors here so that we make sure that offsets are handled concurrently
if (errorHandler != null) {
errorHandler.handle(e, kafkaMessage);
}
}
finally {
if (messageListener != null) {
offsetManager.updateOffset(kafkaMessage.getMetadata().getPartition(),
kafkaMessage.getMetadata().getNextOffset());
}
}
}
@Override
public void onError(Throwable t) {
//ignore
}
@Override
public void onComplete() {
CountDownLatch latch = shutdownLatch;
if (latch != null) {
shutdownLatch = null;
latch.countDown();
}
}
}
}

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerFactoryBean.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.support;
import java.util.Properties;
@@ -28,7 +29,6 @@ import org.springframework.beans.factory.FactoryBean;
/**
* @author Soby Chacko
* @author Marius Bogoevici
*
* @since 0.5
*/
public class ProducerFactoryBean<K, V> implements FactoryBean<Producer<K, V>> {
@@ -42,7 +42,7 @@ public class ProducerFactoryBean<K, V> implements FactoryBean<Producer<K, V>> {
private Properties producerProperties = new Properties();
public ProducerFactoryBean(final ProducerMetadata<K, V> producerMetadata, final String brokerList,
final Properties producerProperties) {
final Properties producerProperties) {
this.producerMetadata = producerMetadata;
this.brokerList = brokerList;
if (producerProperties != null) {
@@ -60,6 +60,7 @@ public class ProducerFactoryBean<K, V> implements FactoryBean<Producer<K, V>> {
props.putAll(producerProperties);
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, producerMetadata.getCompressionType().name());
props.put(ProducerConfig.BATCH_SIZE_CONFIG, producerMetadata.getBatchBytes());
LOGGER.info("Using producer properties => " + props);
return new KafkaProducer<>(props,
producerMetadata.getKeySerializer(),

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

@@ -146,7 +146,6 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
});
Message<?> m = getAMessageFrom(this.kafkaListener, toMessage.get());
assertThat(m.getClass().getSimpleName(), equalTo("KafkaMessage"));
assertNull(m.getHeaders().getId());
assertNull(m.getHeaders().getTimestamp());
assertNull(m.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT));
@@ -167,9 +166,10 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
assertRest(m);
m = getAMessageFrom(this.withOverrideIdTS, toMessage.get());
assertThat(m.getClass().getSimpleName(), equalTo("KafkaMessage"));
assertNotNull(m.getHeaders().getId());
assertNotNull(m.getHeaders().getTimestamp());
//TODO org.springframework.messaging.support.MessageBuilder doesn't support simple way
// to provide TIMESTAMP generation option.
// assertNotNull(m.getHeaders().getTimestamp());
assertNull(m.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT));
assertRest(m);
}