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

Add Support for ProducerListener

Browse Source 
Introduces the concept of ProducerListener, that can be injected in a ProducerContext and receive notifications when a message has been acknowledged or rejected.

While the Kafka API currently supports providing a Callback, it requires the creation of a distinct instance for receiving contextual information about the message that has been sent or rejected. This allows the use of the Callback mechanism with an injectable ProducerListener strategy.

Polishing and Callback Integration Test
This commit is contained in:
Marius Bogoevici
2015-10-24 16:09:09 -04:00
committed by Artem Bilan
parent d8a1115f1f
commit 464ca70aa6
11 changed files with 417 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java
View File

@@ -157,6 +157,8 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars
.addConstructorArgValue(producerFactoryBeanDefinition);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerConfigurationBuilder, producerConfiguration,
"conversion-service");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerConfigurationBuilder, producerConfiguration,
"producer-listener");
producerConfigurationsMap.put(producerConfiguration.getAttribute("topic"),
producerConfigurationBuilder.getBeanDefinition());
}

36
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultProducerListener.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.support;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
* No-op implementation of @link ProducerListener}, to be used as base class for other implementations.
*
* @author Marius Bogoevici
* @since 1.3
*/
public class DefaultProducerListener implements ProducerListener {
@Override
public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
}
@Override
public void onError(String topic, Integer partition, Object key, Object value, Exception exception) {
}
}

83
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/LoggingProducerListener.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.support;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.ObjectUtils;
/**
* {@link ProducerListener} that logs exceptions thrown when sending messages.
*
* @author Marius Bogoevici
* @since 1.3
*/
public class LoggingProducerListener extends DefaultProducerListener {
private static final Log log = LogFactory.getLog(LoggingProducerListener.class);
private boolean includeContents = true;
private int maxContentLogged = 100;
/**
* Whether the log message should include the contents (key and payload).
*
* @param includeContents true if the contents of the message should be logged
*/
public void setIncludeContents(boolean includeContents) {
this.includeContents = includeContents;
}
/**
* The maximum amount of data to be logged for either key or password. As message sizes may vary and
* become fairly large, this allows limiting the amount of data sent to logs.
*
* @param maxContentLogged the maximum amount of data being logged.
*/
public void setMaxContentLogged(int maxContentLogged) {
this.maxContentLogged = maxContentLogged;
}
@Override
public void onError(String topic, Integer partition, Object key, Object payload, Exception exception) {
if (log.isErrorEnabled()) {
StringBuffer logOutput = new StringBuffer();
logOutput.append("Exception thrown when sending a message");
if (includeContents) {
logOutput.append(" with key='"
+ toDisplayString(ObjectUtils.nullSafeToString(key), maxContentLogged) + "'");
logOutput.append(" and payload='"
+ toDisplayString(ObjectUtils.nullSafeToString(payload),maxContentLogged) + "'");
}
logOutput.append(" to topic " + topic);
if (partition != null) {
logOutput.append(" and partition " +partition);
}
logOutput.append(":");
log.error(logOutput, exception);
}
}
private String toDisplayString(String original, int maxCharacters) {
if (original.length() <= maxCharacters) {
return original;
}
return original.substring(0, maxCharacters) + "...";
}
}

19
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java
View File

@@ -49,6 +49,8 @@ public class ProducerConfiguration<K, V> {
private ConversionService conversionService;
private ProducerListener producerListener;
public ProducerConfiguration(ProducerMetadata<K, V> producerMetadata, Producer<K, V> producer) {
Assert.notNull(producerMetadata);
Assert.notNull(producer);
@@ -64,6 +66,10 @@ public class ProducerConfiguration<K, V> {
this.conversionService = conversionService;
}
public void setProducerListener(ProducerListener producerListener) {
this.producerListener = producerListener;
}
public ProducerMetadata<K, V> getProducerMetadata() {
return this.producerMetadata;
}
@@ -85,9 +91,18 @@ public class ProducerConfiguration<K, V> {
partition = this.getProducerMetadata().getPartitioner().partition(messageKey,
this.producer.partitionsFor(targetTopic).size());
}
Future<RecordMetadata> future =
this.producer.send(new ProducerRecord<>(targetTopic, partition, messageKey, messagePayload));
ProducerRecord<K, V> record = new ProducerRecord<>(targetTopic, partition, messageKey, messagePayload);
Future<RecordMetadata> future;
if (producerListener == null) {
future = this.producer.send(record);
}
else {
ProducerListenerInvokingCallback callback =
new ProducerListenerInvokingCallback(targetTopic, partition, messageKey, messagePayload,
producerListener);
future = this.producer.send(record, callback);
}
if (!producerMetadata.isSync()) {
return future;
}

55
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListener.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.support;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
* Listener for handling outbound Kafka messages. Exactly one of its methods will be invoked, depending on whether
* the write has been acknowledged or not.
*
* Its main goal is to provide a stateless singleton delegate for {@link org.apache.kafka.clients.producer.Callback}s,
* which, in all but the most trivial cases, requires creating a separate instance per message.
*
* @see org.apache.kafka.clients.producer.Callback
* @author Marius Bogoevici
* @since 1.3
*/
public interface ProducerListener {
/**
* Invoked after the successful send of a message (that is, after it has been acknowledged by the broker)
*
* @param topic the destination topic
* @param partition the destination partition (could be null)
* @param key the key of the outbound message
* @param value the payload of the outbound message
* @param recordMetadata the result of the successful send operation
*/
void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata);
/**
* Invoked after an attempt to send a message has failed
*
* @param topic the destination topic
* @param partition the destination partition (could be null)
* @param key the key of the outbound message
* @param value the payload of the outbound message
* @param exception the exception thrown
*/
void onError(String topic, Integer partition, Object key, Object value,Exception exception);
}

61
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListenerInvokingCallback.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.support;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.util.Assert;
/**
* Adapts the {@link org.apache.kafka.clients.producer.Callback} interface of the
* {@link org.apache.kafka.clients.producer.Producer} to a {@link ProducerListener}.
*
* @author Marius Bogoevici
*/
public class ProducerListenerInvokingCallback implements Callback {
private final String topic;
private final Integer partition;
private final Object key;
private final Object payload;
private final ProducerListener producerListener;
public ProducerListenerInvokingCallback(String topic, Integer partition, Object key, Object payload,
ProducerListener producerListener) {
Assert.notNull(producerListener, "must not be null");
this.topic = topic;
this.partition = partition;
this.key = key;
this.payload = payload;
this.producerListener = producerListener;
}
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
producerListener.onError(topic,partition,key, payload,exception);
}
else {
producerListener.onSuccess(topic, partition, key, payload, metadata);
}
}
}

12
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd
View File

@@ -189,6 +189,18 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="producer-listener" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Listener for notifying when a message has been sent or has failed.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.kafka.support.ProducerListener"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="compression-type" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[

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

@@ -46,6 +46,7 @@
batch-bytes="9876"
partitioner="partitioner"
conversion-service="conversionService"
producer-listener="producerListener"
compression-type="none"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
@@ -54,6 +55,8 @@
<constructor-arg value="java.lang.String" />
</bean>
<bean id="producerListener" class="org.springframework.integration.kafka.support.LoggingProducerListener"/>
<bean id="stringSerializer" class="org.apache.kafka.common.serialization.StringSerializer"/>
<bean id="partitioner" class="org.springframework.integration.kafka.support.DefaultPartitioner"/>

9
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParserTests.java
View File

@@ -36,9 +36,12 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.integration.kafka.rule.KafkaEmbedded;
import org.springframework.integration.kafka.rule.KafkaRule;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerListener;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
import org.springframework.integration.test.util.TestUtils;
@@ -55,7 +58,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
public class KafkaProducerContextParserTests {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
public static KafkaRule kafkaRule = new KafkaEmbedded(1);
@Autowired
private ApplicationContext appContext;
@@ -108,6 +111,10 @@ public class KafkaProducerContextParserTests {
ConversionService configuredConversionService = (ConversionService) directFieldAccessor2.getPropertyValue("conversionService");
assertSame(conversionService, configuredConversionService);
final ProducerListener producerListener = appContext.getBean("producerListener", ProducerListener.class);
ProducerListener configuredProducerListener = (ProducerListener) directFieldAccessor2.getPropertyValue("producerListener");
assertSame(producerListener, configuredProducerListener);
assertEquals(9876,producerConfigurationTest2.getProducerMetadata().getBatchBytes());
assertFalse(TestUtils.getPropertyValue(producerContext, "autoStartup", Boolean.class));

39
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java
View File

@@ -22,6 +22,7 @@ import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import java.util.ArrayList;
@@ -33,6 +34,7 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
@@ -53,6 +55,7 @@ import org.springframework.integration.kafka.support.KafkaHeaders;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerFactoryBean;
import org.springframework.integration.kafka.support.ProducerListener;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.integration.kafka.util.EncoderAdaptingSerializer;
@@ -89,6 +92,7 @@ public class OutboundTests {
public void tearDown() {
try {
AdminUtils.deleteTopic(kafkaRule.getZkClient(), TOPIC);
AdminUtils.deleteTopic(kafkaRule.getZkClient(), TOPIC2);
}
catch (Exception e) {
}
@@ -156,7 +160,7 @@ public class OutboundTests {
}
@Test
public void testHeaderRouting() throws Exception {
public void testHeaderRoutingAndAsyncCallback() throws Exception {
// create the topic
@@ -196,7 +200,24 @@ public class OutboundTests {
kafkaMessageListenerContainer.start();
KafkaProducerContext producerContext = createProducerContext();
int expectedDeliveryConfirmations = 4;
final List<RecordMetadata> results = new ArrayList<RecordMetadata>();
final CountDownLatch sendResultLatch = new CountDownLatch(expectedDeliveryConfirmations);
ProducerListener listener = new ProducerListener() {
@Override
public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
results.add(recordMetadata);
sendResultLatch.countDown();
}
@Override
public void onError(String topic, Integer partition, Object key, Object value, Exception exception) {
sendResultLatch.countDown();
}
};
KafkaProducerContext producerContext = createProducerContext(listener);
KafkaProducerMessageHandler handler
= new KafkaProducerMessageHandler(producerContext);
@@ -226,9 +247,12 @@ public class OutboundTests {
.setHeader("bar", TOPIC2)
.build());
assertTrue(sendResultLatch.await(10, TimeUnit.SECONDS));
assertThat(results.size(), equalTo(expectedDeliveryConfirmations));
producerContext.stop();
latch.await(1000, TimeUnit.MILLISECONDS);
latch.await(10000, TimeUnit.MILLISECONDS);
assertThat(latch.getCount(), equalTo(0L));
// messages are routed to both topics
assertThat(payloadsByTopic.keysView(), hasItem(TOPIC));
@@ -309,17 +333,24 @@ public class OutboundTests {
}
private KafkaProducerContext createProducerContext() throws Exception {
return createProducerContext(null);
}
private KafkaProducerContext createProducerContext(ProducerListener producerListener) throws Exception {
KafkaProducerContext kafkaProducerContext = new KafkaProducerContext();
Encoder<String> encoder = new StringEncoder();
ProducerMetadata<String, String> producerMetadata =
new ProducerMetadata<String, String>(TOPIC, String.class, String.class,
new EncoderAdaptingSerializer<>(encoder), new EncoderAdaptingSerializer<>(encoder));
Properties props = new Properties();
props.put("linger.ms", "15000");
if (producerListener == null) {
props.put("linger.ms", "15000");
}
ProducerFactoryBean<String, String> producer =
new ProducerFactoryBean<>(producerMetadata, kafkaRule.getBrokersAsString(), props);
ProducerConfiguration<String, String> config =
new ProducerConfiguration<>(producerMetadata, producer.getObject());
config.setProducerListener(producerListener);
Map<String, ProducerConfiguration<?, ?>> producerConfigurationMap =
Collections.<String, ProducerConfiguration<?, ?>>singletonMap(TOPIC, config);
kafkaProducerContext.setProducerConfigurations(producerConfigurationMap);

105
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/ProducerListenerTests.java Normal file
View File

@@ -0,0 +1,105 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.outbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import java.util.Collections;
import java.util.Map;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.Serializer;
import org.hamcrest.CoreMatchers;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.kafka.support.KafkaHeaders;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerListener;
import org.springframework.integration.kafka.support.ProducerListenerInvokingCallback;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Marius Bogoevici
*/
public class ProducerListenerTests {
@Test
@SuppressWarnings({"unchecked","rawtypes"})
public void testProducerListenerSet() throws Exception {
KafkaProducerContext producerContext = new KafkaProducerContext();
Serializer serializer = Mockito.mock(Serializer.class);
ProducerMetadata<Object, Object> producerMetadata
= new ProducerMetadata<>("default", Object.class, Object.class, serializer, serializer);
Producer producer = Mockito.mock(Producer.class);
ProducerConfiguration<Object, Object> producerConfiguration
= new ProducerConfiguration<>(producerMetadata, producer);
ProducerListener producerListener = mock(ProducerListener.class);
producerConfiguration.setProducerListener(producerListener);
Map<String, ProducerConfiguration<?, ?>> producerConfigurations
= Collections.<String, ProducerConfiguration<?, ?>>singletonMap("default", producerConfiguration);
producerContext.setProducerConfigurations(producerConfigurations);
KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(producerContext);
handler.handleMessage(
MessageBuilder.withPayload("somePayload")
.setHeader(KafkaHeaders.PARTITION_ID, 2)
.setHeader(KafkaHeaders.MESSAGE_KEY, "someKey")
.build());
final ArgumentCaptor<Callback> argument = ArgumentCaptor.forClass(Callback.class);
verify(producer).send(any(ProducerRecord.class), argument.capture());
Callback callback = argument.getValue();
assertThat(callback, CoreMatchers.instanceOf(ProducerListenerInvokingCallback.class));
DirectFieldAccessor fieldAccessor = new DirectFieldAccessor(callback);
assertEquals(fieldAccessor.getPropertyValue("topic"),"default");
assertEquals(fieldAccessor.getPropertyValue("partition"),2);
assertEquals(fieldAccessor.getPropertyValue("key"),"someKey");
assertEquals(fieldAccessor.getPropertyValue("payload"),"somePayload");
assertSame(fieldAccessor.getPropertyValue("producerListener"), producerListener);
verifyNoMoreInteractions(producer);
}
@Test
@SuppressWarnings({"unchecked","rawtypes"})
public void testProducerListenerNotSet() throws Exception {
KafkaProducerContext producerContext = new KafkaProducerContext();
Serializer serializer = Mockito.mock(Serializer.class);
ProducerMetadata<Object, Object> producerMetadata
= new ProducerMetadata<>("default", Object.class, Object.class, serializer, serializer);
Producer producer = Mockito.mock(Producer.class);
ProducerConfiguration<Object, Object> producerConfiguration
= new ProducerConfiguration<>(producerMetadata, producer);
Map<String, ProducerConfiguration<?, ?>> producerConfigurations
= Collections.<String, ProducerConfiguration<?, ?>>singletonMap("default", producerConfiguration);
producerContext.setProducerConfigurations(producerConfigurations);
KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(producerContext);
handler.handleMessage(MessageBuilder.withPayload("somePayload").build());
verify(producer).send(any(ProducerRecord.class));
verifyNoMoreInteractions(producer);
}
}