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

IE-100: Kafka:o-c-a: topic and message-key attrs

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

Add `topic(-expression)` and `message-key(-expression)` attributes to avoid upstream configuration
to specify them in the `MessageHeaders`

Polishing for XSD

Conflicts:
	spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java
	spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java
	spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd
	spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java

INTEXT-100: Add `Assert.notNull(this.evaluationContext);`

INTEXT-100: Fix parser potential NPE

INTEXT-100: Add `KafkaHeaders`

Merge branch 'INTEXT-100-1' of ..\spring-integration-extensions into INTEXT-100

Conflicts:
	src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
	src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java
	src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java
	src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java

* Polishing according the rebase to `master`
* Polishing for the `KafkaRunning` Rule to use `ZkClient` and its `getAllBrokersInCluster`
* Add a note to the `README.md`

Addressing PR comments

Polishing

- Minor doc polish + change tabs to spaces in code
- Enhance test to include expressions
This commit is contained in:
Artem Bilan
2014-11-17 13:22:39 +02:00
committed by Artem Bilan
parent 0e2808c014
commit 614308b482
11 changed files with 476 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2013-2014 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.
@@ -15,21 +15,26 @@
*/
package org.springframework.integration.kafka.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
*
* @author Soby Chacko
* @author Artem Bilan
* @since 0.5
*
*/
public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(final Element element, final ParserContext parserContext) {
final BeanDefinitionBuilder kafkaProducerMessageHandlerBuilder =
@@ -41,6 +46,21 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd
kafkaProducerMessageHandlerBuilder.addConstructorArgReference(kafkaServerBeanName);
}
BeanDefinition topicExpressionDef =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("topic", "topic-expression",
parserContext, element, false);
if (topicExpressionDef != null) {
kafkaProducerMessageHandlerBuilder.addPropertyValue("topicExpression", topicExpressionDef);
}
BeanDefinition messageKeyExpressionDef =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("message-key",
"message-key-expression", parserContext, element, false);
if (messageKeyExpressionDef != null) {
kafkaProducerMessageHandlerBuilder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef);
}
return kafkaProducerMessageHandlerBuilder.getBeanDefinition();
}
}

46
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -13,32 +13,72 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.outbound;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.kafka.support.KafkaHeaders;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* @author Soby Chacko
* @author Artem Bilan
* @author Gary Russell
* @since 0.5
*/
public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler {
public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler
implements IntegrationEvaluationContextAware {
private final KafkaProducerContext<K,V> kafkaProducerContext;
private EvaluationContext evaluationContext;
private volatile Expression topicExpression;
private volatile Expression messageKeyExpression;
public KafkaProducerMessageHandler(final KafkaProducerContext<K,V> kafkaProducerContext) {
this.kafkaProducerContext = kafkaProducerContext;
}
@Override
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
this.evaluationContext = evaluationContext;
}
public void setTopicExpression(Expression topicExpression) {
this.topicExpression = topicExpression;
}
public void setMessageKeyExpression(Expression messageKeyExpression) {
this.messageKeyExpression = messageKeyExpression;
}
public KafkaProducerContext<K,V> getKafkaProducerContext() {
return kafkaProducerContext;
return this.kafkaProducerContext;
}
@Override
protected void onInit() throws Exception {
Assert.notNull(this.evaluationContext);
}
@Override
protected void handleMessageInternal(final Message<?> message) throws Exception {
kafkaProducerContext.send(message);
String topic = this.topicExpression != null
? this.topicExpression.getValue(this.evaluationContext, message, String.class)
: message.getHeaders().get(KafkaHeaders.TOPIC, String.class);
Object messageKey = this.messageKeyExpression != null
? this.messageKeyExpression.getValue(this.evaluationContext, message)
: message.getHeaders().get(KafkaHeaders.MESSAGE_KEY);
this.kafkaProducerContext.send(topic, messageKey, message);
}
@Override

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

@@ -0,0 +1,31 @@
/*
* Copyright 2014 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;
/**
* @author Artem Bilan
* @since 1.0
*/
public abstract class KafkaHeaders {
private static final String PREFIX = "kafka_";
public static final String TOPIC = PREFIX + "topic";
public static final String MESSAGE_KEY = PREFIX + "messageKey";
}

33
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.kafka.support;
import java.util.Collection;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
@@ -34,6 +33,7 @@ import org.springframework.messaging.Message;
* @author Rajasekar Elango
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @author Artem Bilan
* @since 0.5
*/
public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponent, BeanNameAware {
@@ -46,8 +46,6 @@ public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponen
private volatile ProducerConfiguration<K, V> theProducerConfiguration;
private Properties producerProperties;
private String beanName = "not_specified";
private int phase = 0;
@@ -83,21 +81,6 @@ public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponen
}
}
/**
* @param producerProperties
* The producerProperties to set.
*/
public void setProducerProperties(Properties producerProperties) {
this.producerProperties = producerProperties;
}
/**
* @return the producerProperties.
*/
public Properties getProducerProperties() {
return this.producerProperties;
}
/**
* @return the component type.
* @since 1.0
@@ -199,21 +182,19 @@ public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponen
callback.run();
}
public void send(final Message<?> message) throws Exception {
public void send(String topic, Object messageKey, final Message<?> message) throws Exception {
if (!running.get()) {
start();
}
if (message.getHeaders().containsKey("topic")) {
ProducerConfiguration<K, V> producerConfiguration =
getTopicConfiguration(message.getHeaders().get("topic", String.class));
if (producerConfiguration != null) {
producerConfiguration.send(message);
}
ProducerConfiguration<K,V> producerConfiguration = getTopicConfiguration(topic);
if (producerConfiguration != null) {
producerConfiguration.send(topic, messageKey, message);
}
// if there is a single producer configuration then use that config to send message.
else if (this.theProducerConfiguration != null) {
this.theProducerConfiguration.send(message);
this.theProducerConfiguration.send(null, messageKey, message);
}
else {
throw new IllegalStateException("Could not send messages as there are multiple producer configurations " +

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

@@ -26,6 +26,7 @@ import org.apache.commons.lang.builder.HashCodeBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
@@ -59,19 +60,14 @@ public class ProducerConfiguration<K, V> {
return this.producer;
}
public void send(final Message<?> message) throws Exception {
public void send(String topic, Object messageKey, final Message<?> message) throws Exception {
final V v = getPayload(message);
String topic = message.getHeaders().containsKey("topic")
? message.getHeaders().get("topic", String.class)
: this.producerMetadata.getTopic();
if (!StringUtils.hasText(topic)) {
topic = this.producerMetadata.getTopic();
}
if (message.getHeaders().containsKey("messageKey")) {
this.producer.send(new KeyedMessage<K, V>(topic, getKey(message), v));
}
else {
this.producer.send(new KeyedMessage<K, V>(topic, v));
}
this.producer.send(new KeyedMessage<K, V>(topic, (messageKey != null ? getKey(messageKey) : null), v));
}
@SuppressWarnings("unchecked")
@@ -86,13 +82,12 @@ public class ProducerConfiguration<K, V> {
}
@SuppressWarnings("unchecked")
private K getKey(final Message<?> message) throws Exception {
final Object key = message.getHeaders().get("messageKey");
private K getKey(Object messageKey) throws Exception {
if (this.producerMetadata.getKeyEncoder() instanceof DefaultEncoder) {
return (K) getByteStream(key);
return (K) getByteStream(messageKey);
}
return message.getHeaders().get("messageKey", this.producerMetadata.getKeyClassType());
return (K) messageKey;
}
private static boolean isRawByteArray(final Object obj) {

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

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/kafka"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/kafka"
@@ -26,16 +26,11 @@
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="zk-connect" use="optional">
<xsd:attribute name="zk-connect" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the Kafka server URL
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="zk-connection-timeout" use="optional">
@@ -43,48 +38,31 @@
<xsd:documentation><![CDATA[
Indicates the Kafka consumer zkConnectionTimeout value
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.Integer"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:int xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="zk-session-timeout" use="optional">
<xsd:attribute name="zk-session-timeout" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the Kafka consumer group id
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="zk-sync-time" use="optional">
<xsd:attribute name="zk-sync-time" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the Kafka consumer group id
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-commit-interval" use="optional">
<xsd:attribute name="auto-commit-interval" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the Kafka consumer group id
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
@@ -120,79 +98,91 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="broker-list" use="required">
<xsd:attribute name="broker-list" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
list of comma separated kafka brokers.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-encoder" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Custom implementation of a Kafka Encoder for encoding message values.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-encoder" use="optional"
type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Custom implementation of a Kafka Encoder for encoding message keys.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-class-type" use="optional"
type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Class type used for the key
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-class-type" use="optional"
type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Class type used for the value
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="compression-codec" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the type of compression codec used for message compression.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
<xsd:documentation>
Custom implementation of a Kafka Encoder for encoding message values.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Encoder"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="partitioner" use="optional"
type="xsd:string">
<xsd:attribute name="key-encoder" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Custom Kafka key partitioner.
</xsd:documentation>
<xsd:appinfo>
<xsd:documentation>
Custom implementation of a Kafka Encoder for encoding message keys.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Encoder"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="async" use="optional"
type="xsd:boolean">
<xsd:attribute name="key-class-type" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Class type used for the key
</xsd:documentation>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.Class"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-class-type" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Class type used for the value
</xsd:documentation>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.Class"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="compression-codec" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the type of compression codec used for message compression.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="partitioner" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Custom Kafka key partitioner.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.producer.Partitioner"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="async" use="optional">
<xsd:annotation>
<xsd:documentation>
Indicates if this producer is async or not.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="batch-num-messages" use="optional"
type="xsd:string">
<xsd:attribute name="batch-num-messages" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
number of messages to batch at this producer.
@@ -206,12 +196,16 @@
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="producer-properties" use="optional"
type="xsd:string">
<xsd:attribute name="producer-properties" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka producer properties to use for all producers
</xsd:documentation>
<xsd:appinfo>
<xsd:documentation>
Kafka producer properties to use for all producers
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.Properties"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup" />
@@ -252,14 +246,9 @@
<xsd:complexType>
<xsd:attribute name="pattern" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
Regex pattern to match topic
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Regex pattern to match topic
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="streams" type="xsd:string" use="required">
@@ -267,11 +256,6 @@
<xsd:documentation><![CDATA[
Number of streams (threads) to use to consume messages
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exclude" use="optional" default="false">
@@ -281,11 +265,6 @@
If exclude is true, it uses blacklist to exclude topics matching given pattern.
Default value is false.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
@@ -294,44 +273,42 @@
</xsd:complexType>
</xsd:element>
</xsd:choice>
<xsd:attribute name="max-messages" use="optional">
<xsd:attribute name="max-messages" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates max messages to aggregate in a single call to receive
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="group-id" use="required">
<xsd:attribute name="group-id" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the Kafka consumer group id
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-decoder" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
<xsd:documentation>
Custom implementation of a Kafka Decoder for values.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Decoder"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="value-decoder" use="optional"
type="xsd:string">
<xsd:attribute name="key-decoder" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka Server Bean Name
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-decoder" use="optional"
type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka Server Bean Name
</xsd:documentation>
<xsd:appinfo>
<xsd:documentation>
Custom implementation of a Kafka Decoder for keys.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Decoder"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
@@ -342,32 +319,30 @@
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="required"/>
<xsd:attribute name="consumer-timeout" use="optional">
<xsd:attribute name="consumer-timeout" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the Kafka consumer timeout ms
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="zookeeper-connect" use="required"
type="xsd:string">
<xsd:attribute name="zookeeper-connect" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka Server Bean Name
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="consumer-properties" use="optional"
type="xsd:string">
<xsd:attribute name="consumer-properties" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka consumer properties to use for all consumers
</xsd:documentation>
<xsd:appinfo>
<xsd:documentation>
Kafka consumer properties to use for all consumers
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.Properties"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
@@ -402,23 +377,23 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="group-id" use="optional">
<xsd:attribute name="group-id" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Indicates the Kafka consumer group id
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="direct">
<tool:expected-type type="java.lang.String"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="kafka-consumer-context-ref" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka Server Bean Name
</xsd:documentation>
<xsd:appinfo>
<xsd:documentation>
Kafka consumer context reference.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.kafka.support.KafkaConsumerContext"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
@@ -439,9 +414,48 @@
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="kafka-producer-context-ref" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka producer context reference.
</xsd:documentation>
<xsd:appinfo>
<xsd:documentation>
Kafka producer context reference.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.kafka.support.KafkaProducerContext"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="topic" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the name of the Kafka topic.
This attribute is mutually exclusive with 'topic-expression' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="topic-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the name of the Kafka topic
against the Message at runtime.
This attribute is mutually exclusive with 'topic' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-key" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the Key for the Kafka message.
This attribute is mutually exclusive with 'message-key-expression' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-key-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the Key for Kafka message
against the Message at runtime.
This attribute is mutually exclusive with 'message-key' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
@@ -451,7 +465,11 @@
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:int xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>

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

@@ -1,45 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
<int:channel id="inputToKafka">
<int:queue/>
</int:channel>
<int:channel id="inputToKafka">
<int:queue/>
</int:channel>
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
kafka-producer-context-ref="kafkaProducerContext"
auto-startup="false"
channel="inputToKafka"
order="3"
>
<int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
</int-kafka:outbound-channel-adapter>
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
kafka-producer-context-ref="kafkaProducerContext"
auto-startup="false"
channel="inputToKafka"
order="3"
topic="foo"
message-key-expression="'bar'">
<int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
</int-kafka:outbound-channel-adapter>
<task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
<task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
<constructor-arg value="java.lang.String" />
</bean>
<bean id="kafkaEncoder"
class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder">
<constructor-arg value="java.lang.String"/>
</bean>
<int-kafka:producer-context id="kafkaProducerContext">
<int-kafka:producer-configurations>
<int-kafka:producer-configuration broker-list="localhost:9092"
key-class-type="java.lang.String"
value-class-type="java.lang.String"
topic="test1"
value-encoder="kafkaEncoder"
key-encoder="kafkaEncoder"
compression-codec="default"/>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="test2"
compression-codec="default"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
<int-kafka:producer-context id="kafkaProducerContext">
<int-kafka:producer-configurations>
<int-kafka:producer-configuration broker-list="localhost:9092"
key-class-type="java.lang.String"
value-class-type="java.lang.String"
topic="test1"
value-encoder="kafkaEncoder"
key-encoder="kafkaEncoder"
compression-codec="default"/>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="test2"
compression-codec="default"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
</beans>

26
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java
View File

@@ -13,9 +13,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.config.xml;
import org.junit.Assert;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -26,11 +29,13 @@ import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Soby Chacko
* @author Artem Bilan
* @author Gary Russell
* @since 0.5
*/
@@ -47,15 +52,16 @@ public class KafkaOutboundAdapterParserTests<K, V> {
@Test
@SuppressWarnings("unchecked")
public void testOutboundAdapterConfiguration() {
final PollingConsumer pollingConsumer =
appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
final KafkaProducerMessageHandler<K, V> messageHandler = appContext.getBean(KafkaProducerMessageHandler.class);
Assert.assertNotNull(pollingConsumer);
Assert.assertNotNull(messageHandler);
Assert.assertEquals(messageHandler.getOrder(), 3);
final KafkaProducerContext<K, V> producerContext = messageHandler.getKafkaProducerContext();
Assert.assertNotNull(producerContext);
Assert.assertEquals(producerContext.getProducerConfigurations().size(), 2);
PollingConsumer pollingConsumer = this.appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
KafkaProducerMessageHandler<K, V> messageHandler = this.appContext.getBean(KafkaProducerMessageHandler.class);
assertNotNull(pollingConsumer);
assertNotNull(messageHandler);
assertEquals(messageHandler.getOrder(), 3);
assertEquals("foo", TestUtils.getPropertyValue(messageHandler, "topicExpression.literalValue"));
assertEquals("'bar'", TestUtils.getPropertyValue(messageHandler, "messageKeyExpression.expression"));
KafkaProducerContext<K, V> producerContext = messageHandler.getKafkaProducerContext();
assertNotNull(producerContext);
assertEquals(producerContext.getProducerConfigurations().size(), 2);
}
}

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

@@ -23,9 +23,18 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.admin.AdminUtils;
import kafka.consumer.ConsumerConfig;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
import org.junit.AfterClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.context.expression.MapAccessor;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.serializer.common.StringDecoder;
import org.springframework.integration.kafka.serializer.common.StringEncoder;
@@ -34,6 +43,7 @@ import org.springframework.integration.kafka.support.ConsumerConfiguration;
import org.springframework.integration.kafka.support.ConsumerConnectionProvider;
import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.integration.kafka.support.KafkaHeaders;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.MessageLeftOverTracker;
import org.springframework.integration.kafka.support.ProducerConfiguration;
@@ -43,10 +53,6 @@ import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import kafka.consumer.ConsumerConfig;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
/**
* @author Gary Russell
* @since 1.0
@@ -59,6 +65,15 @@ public class OutboundTests {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@AfterClass
public static void tearDown() {
try {
AdminUtils.deleteTopic(kafkaRunning.getZkClient(), TOPIC);
}
catch (Exception e) {
}
}
@Test
public void testAsyncProducerFlushed() throws Exception {
KafkaConsumerContext<String, String> consumerContext = createConsumer();
@@ -83,14 +98,29 @@ public class OutboundTests {
KafkaProducerMessageHandler<String, String> handler = new KafkaProducerMessageHandler<String, String>(kafkaProducerContext);
handler.handleMessage(MessageBuilder.withPayload("foo")
.setHeader("messagekey", "3")
.setHeader("topic", TOPIC)
.setHeader(KafkaHeaders.MESSAGE_KEY, "3")
.setHeader(KafkaHeaders.TOPIC, TOPIC)
.build());
SpelExpressionParser parser = new SpelExpressionParser();
handler.setMessageKeyExpression(parser.parseExpression("headers.foo"));
handler.setTopicExpression(parser.parseExpression("headers.bar"));
StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
evaluationContext.addPropertyAccessor(new MapAccessor());
handler.setIntegrationEvaluationContext(evaluationContext);
handler.handleMessage(MessageBuilder.withPayload("bar")
.setHeader("foo", "3")
.setHeader("bar", TOPIC)
.build());
kafkaProducerContext.stop();
Message<Map<String, Map<Integer, List<Object>>>> received = consumerContext.receive();
assertNotNull(received);
if (((Map<?,?>) received.getPayload()).size() < 2) {
received = consumerContext.receive();
assertNotNull(received);
}
consumerContext.destroy();
}

50
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java
View File

@@ -15,11 +15,7 @@
*/
package org.springframework.integration.kafka.rule;
import java.io.IOException;
import java.net.Socket;
import javax.net.SocketFactory;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Assume;
@@ -27,6 +23,11 @@ import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import org.springframework.integration.kafka.core.ZookeeperConnectDefaults;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
/**
* <p>
* A rule that prevents integration tests from failing if the Kafka server is not running or not
@@ -44,9 +45,7 @@ import org.junit.runners.model.Statement;
*/
public class KafkaRunning extends TestWatcher {
public static final int KAFKA_PORT = 9092;
public static final int ZOOKEEPER_PORT = 2181;
private static final String ZOOKEEPER_CONNECT_STRING = ZookeeperConnectDefaults.ZK_CONNECT;
private static final Log logger = LogFactory.getLog(KafkaRunning.class);
@@ -57,37 +56,24 @@ public class KafkaRunning extends TestWatcher {
return new KafkaRunning();
}
private ZkClient zkClient;
public ZkClient getZkClient() {
return zkClient;
}
@Override
public Statement apply(Statement base, Description description) {
Socket kSocket = null;
Socket zSocket = null;
try {
kSocket = SocketFactory.getDefault().createSocket("localhost", KAFKA_PORT);
kSocket.getInputStream();
zSocket = SocketFactory.getDefault().createSocket("localhost", ZOOKEEPER_PORT);
zSocket.getInputStream();
this.zkClient = new ZkClient(ZOOKEEPER_CONNECT_STRING, 1000, 1000, ZKStringSerializer$.MODULE$);
if (ZkUtils.getAllBrokersInCluster(zkClient).size() == 0) {
throw new IllegalStateException("No running Kafka brokers");
}
}
catch (final Exception e) {
catch (Exception e) {
logger.warn("Not executing tests because basic connectivity test failed");
Assume.assumeNoException(e);
}
finally {
if (kSocket != null) {
try {
kSocket.close();
}
catch (IOException e) {
}
}
if (zSocket != null) {
try {
zSocket.close();
}
catch (IOException e) {
}
}
}
return super.apply(base, description);
}

175
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerConfigurationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2013-2014 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.
@@ -13,33 +13,38 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.support;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.serializer.DefaultEncoder;
import kafka.serializer.StringEncoder;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder;
import org.springframework.integration.kafka.test.utils.NonSerializableTestKey;
import org.springframework.integration.kafka.test.utils.NonSerializableTestPayload;
import org.springframework.integration.kafka.test.utils.TestKey;
import org.springframework.integration.kafka.test.utils.TestPayload;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
package org.springframework.integration.kafka.support;
import java.io.ByteArrayInputStream;
import java.io.NotSerializableException;
import java.io.ObjectInputStream;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder;
import org.springframework.integration.kafka.test.utils.NonSerializableTestKey;
import org.springframework.integration.kafka.test.utils.NonSerializableTestPayload;
import org.springframework.integration.kafka.test.utils.TestKey;
import org.springframework.integration.kafka.test.utils.TestPayload;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.serializer.DefaultEncoder;
import kafka.serializer.StringEncoder;
/**
* @author Soby Chacko
* @author Artem Bilan
* @since 0.5
*/
public class ProducerConfigurationTests<K,V> {
public class ProducerConfigurationTests<K, V> {
@Test
@SuppressWarnings("unchecked")
public void testSendMessageWithNonDefaultKeyAndValueEncoders() throws Exception {
@@ -50,20 +55,16 @@ public class ProducerConfigurationTests<K,V> {
producerMetadata.setValueClassType(String.class);
final Producer<String, String> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<String, String> configuration = new ProducerConfiguration<String, String>(producerMetadata, producer);
final ProducerConfiguration<String, String> configuration =
new ProducerConfiguration<String, String>(producerMetadata, producer);
final Message<String> message = MessageBuilder.withPayload("test message")
.setHeader("messageKey", "key")
.setHeader("topic", "test")
.build();
configuration.send(message);
configuration.send("test", "key", new GenericMessage<String>("test message"));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
final ArgumentCaptor<KeyedMessage<String, String>> argument =
(ArgumentCaptor<KeyedMessage<String, String>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
ArgumentCaptor.forClass(KeyedMessage.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<String, String> capturedKeyMessage = argument.getValue();
@@ -78,48 +79,47 @@ public class ProducerConfigurationTests<K,V> {
*/
@Test
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersAndCustomSerializableKeyAndPayloadObject() throws Exception {
public void testSendMessageWithDefaultKeyAndValueEncodersAndCustomSerializableKeyAndPayloadObject()
throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration = new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final Message<TestPayload> message = MessageBuilder.withPayload(new TestPayload("part1", "part2"))
.setHeader("messageKey", new TestKey("compositePart1", "compositePart2"))
.setHeader("topic", "test")
.build();
Message<TestPayload> message = new GenericMessage<TestPayload>(new TestPayload("part1", "part2"));
configuration.send(message);
configuration.send("test", new TestKey("compositePart1", "compositePart2"), message);
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
final ArgumentCaptor<KeyedMessage<byte[], byte[]>> argument =
(ArgumentCaptor<KeyedMessage<byte[], byte[]>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
ArgumentCaptor.forClass(KeyedMessage.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<byte[], byte[]> capturedKeyMessage = argument.getValue();
final byte[] keyBytes = capturedKeyMessage.key();
final ByteArrayInputStream keyInputStream = new ByteArrayInputStream (keyBytes);
final ObjectInputStream keyObjectInputStream = new ObjectInputStream (keyInputStream);
final ByteArrayInputStream keyInputStream = new ByteArrayInputStream(keyBytes);
final ObjectInputStream keyObjectInputStream = new ObjectInputStream(keyInputStream);
final Object keyObj = keyObjectInputStream.readObject();
final TestKey tk = (TestKey)keyObj;
final TestKey tk = (TestKey) keyObj;
Assert.assertEquals(tk.getKeyPart1(), "compositePart1");
Assert.assertEquals(tk.getKeyPart2(), "compositePart2");
final byte[] messageBytes = capturedKeyMessage.message();
final ByteArrayInputStream messageInputStream = new ByteArrayInputStream (messageBytes);
final ObjectInputStream messageObjectInputStream = new ObjectInputStream (messageInputStream);
final ByteArrayInputStream messageInputStream = new ByteArrayInputStream(messageBytes);
final ObjectInputStream messageObjectInputStream = new ObjectInputStream(messageInputStream);
final Object messageObj = messageObjectInputStream.readObject();
final TestPayload tp = (TestPayload)messageObj;
final TestPayload tp = (TestPayload) messageObj;
Assert.assertEquals(tp.getPart1(), "part1");
Assert.assertEquals(tp.getPart2(), "part2");
@@ -134,34 +134,32 @@ public class ProducerConfigurationTests<K,V> {
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyEncoderAndNonDefaultValueEncoderAndCorrespondingData() throws Exception {
final ProducerMetadata<byte[], TestPayload> producerMetadata = new ProducerMetadata<byte[], TestPayload>("test");
final AvroReflectDatumBackedKafkaEncoder<TestPayload> encoder = new AvroReflectDatumBackedKafkaEncoder<TestPayload>(TestPayload.class);
final AvroReflectDatumBackedKafkaEncoder<TestPayload> encoder =
new AvroReflectDatumBackedKafkaEncoder<TestPayload>(TestPayload.class);
producerMetadata.setValueEncoder(encoder);
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
producerMetadata.setValueClassType(TestPayload.class);
final Producer<byte[], TestPayload> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], TestPayload> configuration = new ProducerConfiguration<byte[], TestPayload>(producerMetadata, producer);
final TestPayload tp = new TestPayload("part1", "part2");
final Message<TestPayload> message = MessageBuilder.withPayload(tp)
.setHeader("messageKey", "key")
.setHeader("topic", "test")
.build();
final ProducerConfiguration<byte[], TestPayload> configuration =
new ProducerConfiguration<byte[], TestPayload>(producerMetadata, producer);
configuration.send(message);
TestPayload tp = new TestPayload("part1", "part2");
configuration.send("test", "key", new GenericMessage<TestPayload>(tp));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
final ArgumentCaptor<KeyedMessage<byte[], TestPayload>> argument =
(ArgumentCaptor<KeyedMessage<byte[], TestPayload>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
ArgumentCaptor.forClass(KeyedMessage.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<byte[], TestPayload> capturedKeyMessage = argument.getValue();
final byte[] keyBytes = capturedKeyMessage.key();
final ByteArrayInputStream keyInputStream = new ByteArrayInputStream (keyBytes);
final ObjectInputStream keyObjectInputStream = new ObjectInputStream (keyInputStream);
final ByteArrayInputStream keyInputStream = new ByteArrayInputStream(keyBytes);
final ObjectInputStream keyObjectInputStream = new ObjectInputStream(keyInputStream);
final Object keyObj = keyObjectInputStream.readObject();
Assert.assertEquals("key", keyObj);
@@ -183,19 +181,18 @@ public class ProducerConfigurationTests<K,V> {
producerMetadata.setKeyClassType(TestKey.class);
final Producer<TestKey, byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<TestKey, byte[]> configuration = new ProducerConfiguration<TestKey, byte[]>(producerMetadata, producer);
final TestKey tk = new TestKey("part1", "part2");
final Message<String> message = MessageBuilder.withPayload("test message").
setHeader("messageKey", tk)
.setHeader("topic", "test").build();
final ProducerConfiguration<TestKey, byte[]> configuration =
new ProducerConfiguration<TestKey, byte[]>(producerMetadata, producer);
configuration.send(message);
final TestKey tk = new TestKey("part1", "part2");
configuration.send("test", tk, new GenericMessage<String>("test message"));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
final ArgumentCaptor<KeyedMessage<TestKey, byte[]>> argument =
(ArgumentCaptor<KeyedMessage<TestKey, byte[]>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
ArgumentCaptor.forClass(KeyedMessage.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<TestKey, byte[]> capturedKeyMessage = argument.getValue();
@@ -204,8 +201,8 @@ public class ProducerConfigurationTests<K,V> {
final byte[] payloadBytes = capturedKeyMessage.message();
final ByteArrayInputStream payloadBis = new ByteArrayInputStream (payloadBytes);
final ObjectInputStream payloadOis = new ObjectInputStream (payloadBis);
final ByteArrayInputStream payloadBis = new ByteArrayInputStream(payloadBytes);
final ObjectInputStream payloadOis = new ObjectInputStream(payloadBis);
final Object payloadObj = payloadOis.readObject();
Assert.assertEquals("test message", payloadObj);
@@ -224,34 +221,31 @@ public class ProducerConfigurationTests<K,V> {
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration = new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final Message<String> message = MessageBuilder.withPayload("test message").
setHeader("messageKey", "key")
.setHeader("topic", "test").build();
configuration.send(message);
configuration.send("test", "key", new GenericMessage<String>("test message"));
Mockito.verify(producer, Mockito.times(1)).send(Mockito.any(KeyedMessage.class));
final ArgumentCaptor<KeyedMessage<byte[], byte[]>> argument =
(ArgumentCaptor<KeyedMessage<byte[], byte[]>>) (Object)
ArgumentCaptor.forClass(KeyedMessage.class);
ArgumentCaptor.forClass(KeyedMessage.class);
Mockito.verify(producer).send(argument.capture());
final KeyedMessage<byte[], byte[]> capturedKeyMessage = argument.getValue();
final byte[] keyBytes = capturedKeyMessage.key();
final ByteArrayInputStream keyBis = new ByteArrayInputStream (keyBytes);
final ObjectInputStream keyOis = new ObjectInputStream (keyBis);
final ByteArrayInputStream keyBis = new ByteArrayInputStream(keyBytes);
final ObjectInputStream keyOis = new ObjectInputStream(keyBis);
final Object keyObj = keyOis.readObject();
Assert.assertEquals("key", keyObj);
final byte[] payloadBytes = capturedKeyMessage.message();
final ByteArrayInputStream payloadBis = new ByteArrayInputStream (payloadBytes);
final ObjectInputStream payloadOis = new ObjectInputStream (payloadBis);
final ByteArrayInputStream payloadBis = new ByteArrayInputStream(payloadBytes);
final ObjectInputStream payloadOis = new ObjectInputStream(payloadBis);
final Object payloadObj = payloadOis.readObject();
Assert.assertEquals("test message", payloadObj);
@@ -269,12 +263,13 @@ public class ProducerConfigurationTests<K,V> {
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration = new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final Message<NonSerializableTestPayload> message = MessageBuilder.withPayload(new NonSerializableTestPayload("part1", "part2")).
setHeader("messageKey", new NonSerializableTestKey("compositePart1", "compositePart2"))
.setHeader("topic", "test").build();
configuration.send(message);
Message<NonSerializableTestPayload> message =
new GenericMessage<NonSerializableTestPayload>(new NonSerializableTestPayload("part1", "part2"));
configuration.send("test", new NonSerializableTestKey("compositePart1", "compositePart2"), message);
}
/**
@@ -282,18 +277,19 @@ public class ProducerConfigurationTests<K,V> {
*/
@Test(expected = NotSerializableException.class)
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersButNonSerializableKeyAndSerializableValue() throws Exception {
public void testSendMessageWithDefaultKeyAndValueEncodersButNonSerializableKeyAndSerializableValue()
throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration = new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final Message<TestPayload> message = MessageBuilder.withPayload(new TestPayload("part1", "part2")).
setHeader("messageKey", new NonSerializableTestKey("compositePart1", "compositePart2"))
.setHeader("topic", "test").build();
configuration.send(message);
Message<TestPayload> message = new GenericMessage<TestPayload>(new TestPayload("part1", "part2"));
configuration.send("test", new NonSerializableTestKey("compositePart1", "compositePart2"), message);
}
/**
@@ -301,17 +297,20 @@ public class ProducerConfigurationTests<K,V> {
*/
@Test(expected = NotSerializableException.class)
@SuppressWarnings("unchecked")
public void testSendMessageWithDefaultKeyAndValueEncodersButSerializableKeyAndNonSerializableValue() throws Exception {
public void testSendMessageWithDefaultKeyAndValueEncodersButSerializableKeyAndNonSerializableValue()
throws Exception {
final ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<byte[], byte[]>("test");
producerMetadata.setValueEncoder(new DefaultEncoder(null));
producerMetadata.setKeyEncoder(new DefaultEncoder(null));
final Producer<byte[], byte[]> producer = Mockito.mock(Producer.class);
final ProducerConfiguration<byte[], byte[]> configuration = new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final ProducerConfiguration<byte[], byte[]> configuration =
new ProducerConfiguration<byte[], byte[]>(producerMetadata, producer);
final Message<NonSerializableTestPayload> message = MessageBuilder.withPayload(new NonSerializableTestPayload("part1", "part2")).
setHeader("messageKey", new TestKey("compositePart1", "compositePart2"))
.setHeader("topic", "test").build();
configuration.send(message);
Message<NonSerializableTestPayload> message =
new GenericMessage<NonSerializableTestPayload>(new NonSerializableTestPayload("part1", "part2"));
configuration.send("test", new TestKey("compositePart1", "compositePart2"), message);
}
}