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

Add CheckStyle, travis; Fix Outbound, Add Tests

Browse Source 
Polishing - PR Comments

Regenerate Gradle `wrapper`
This commit is contained in:
Gary Russell
2016-03-30 14:35:41 -04:00
committed by Artem Bilan
parent a030be3d7e
commit 93e81f74c8
17 changed files with 433 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-kafka/src/checkstyle/checkstyle-header.txt Normal file
View File

@@ -0,0 +1,17 @@
^\Q/*\E$
^\Q * Copyright \E20\d\d(\-20\d\d)?\Q the original author or authors.\E$
^\Q *\E$
^\Q * Licensed under the Apache License, Version 2.0 (the "License");\E$
^\Q * you may not use this file except in compliance with the License.\E$
^\Q * You may obtain a copy of the License at\E$
^\Q *\E$
^\Q * http://www.apache.org/licenses/LICENSE-2.0\E$
^\Q *\E$
^\Q * Unless required by applicable law or agreed to in writing, software\E$
^\Q * distributed under the License is distributed on an "AS IS" BASIS,\E$
^\Q * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\E$
^\Q * See the License for the specific language governing permissions and\E$
^\Q * limitations under the License.\E$
^\Q */\E$
^$
^.*$

9
spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml Normal file
View File

@@ -0,0 +1,9 @@
<?xml version="1.0"?>
<!DOCTYPE suppressions PUBLIC
"-//Puppy Crawl//DTD Suppressions 1.1//EN"
"http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
<suppressions>
<suppress files="package-info\.java" checks=".*" />
<suppress files="[\\/]test[\\/]" checks="RequireThis" />
<suppress files="[\\/]test[\\/]" checks="Javadoc*" />
</suppressions>

181
spring-integration-kafka/src/checkstyle/checkstyle.xml Normal file
View File

@@ -0,0 +1,181 @@
<?xml version="1.0"?>
<!DOCTYPE module PUBLIC "-//Puppy Crawl//DTD Check Configuration 1.2//EN" "http://www.puppycrawl.com/dtds/configuration_1_2.dtd">
<module name="Checker">
<module name="SuppressionFilter">
<property name="file" value="src/checkstyle/checkstyle-suppressions.xml" />
</module>
<!-- Root Checks -->
<module name="RegexpHeader">
<property name="headerFile" value="src/checkstyle/checkstyle-header.txt" />
<property name="fileExtensions" value="java" />
</module>
<module name="NewlineAtEndOfFile">
<property name="lineSeparator" value="lf"/>
</module>
<!-- TreeWalker Checks -->
<module name="TreeWalker">
<!-- Annotations -->
<module name="AnnotationUseStyle">
<property name="elementStyle" value="compact" />
</module>
<module name="MissingOverride" />
<module name="PackageAnnotation" />
<module name="AnnotationLocation">
<property name="allowSamelineSingleParameterlessAnnotation"
value="false" />
</module>
<!-- Block Checks -->
<module name="EmptyBlock">
<property name="option" value="text" />
</module>
<module name="LeftCurly" />
<module name="RightCurly">
<property name="option" value="alone" />
</module>
<module name="NeedBraces" />
<module name="AvoidNestedBlocks" />
<!-- Class Design -->
<module name="FinalClass" />
<module name="InterfaceIsType" />
<module name="HideUtilityClassConstructor" />
<module name="MutableException" />
<module name="InnerTypeLast" />
<module name="OneTopLevelClass" />
<!-- Coding -->
<module name="CovariantEquals" />
<module name="EmptyStatement" />
<module name="EqualsHashCode" />
<module name="InnerAssignment" />
<module name="SimplifyBooleanExpression" />
<module name="SimplifyBooleanReturn" />
<module name="StringLiteralEquality" />
<module name="NestedForDepth">
<property name="max" value="3" />
</module>
<module name="NestedIfDepth">
<property name="max" value="4" />
</module>
<module name="NestedTryDepth">
<property name="max" value="3" />
</module>
<module name="MultipleVariableDeclarations" />
<module name="RequireThis">
<property name="checkMethods" value="false" />
</module>
<module name="OneStatementPerLine" />
<!-- Imports -->
<module name="AvoidStarImport" />
<module name="AvoidStaticImport">
<property name="excludes"
value="org.assertj.core.api.Assertions.*,
org.junit.Assert.*,
org.junit.Assume.*,
org.junit.internal.matchers.ThrowableMessageMatcher.*,
org.hamcrest.CoreMatchers.*,
org.hamcrest.Matchers.*,
org.mockito.Mockito.*,
org.mockito.BDDMockito.*,
org.mockito.Matchers.*,
org.springframework.kafka.test.hamcrest.KafkaMatchers.*,
org.springframework.kafka.test.assertj.KafkaConditions.*" />
</module>
<module name="IllegalImport" />
<module name="RedundantImport" />
<module name="UnusedImports">
<property name="processJavadoc" value="true" />
</module>
<module name="ImportOrder">
<property name="groups" value="java,/^javax?\./,org,org.springframework,*" />
<property name="ordered" value="true" />
<property name="separated" value="true" />
<property name="option" value="top" />
<property name="sortStaticImportsAlphabetically" value="true" />
</module>
<!-- Javadoc Comments -->
<module name="JavadocType">
<property name="scope" value="package"/>
<property name="authorFormat" value=".+\s.+"/>
</module>
<module name="JavadocMethod">
<property name="allowMissingJavadoc" value="true" />
</module>
<module name="JavadocVariable">
<property name="scope" value="public"/>
</module>
<module name="JavadocStyle">
<property name="checkEmptyJavadoc" value="true"/>
</module>
<module name="NonEmptyAtclauseDescription" />
<module name="JavadocTagContinuationIndentation">
<property name="offset" value="0"/>
</module>
<module name="AtclauseOrder">
<property name="target" value="CLASS_DEF, INTERFACE_DEF, ENUM_DEF"/>
<property name="tagOrder" value="@param, @author, @since, @see, @version, @serial, @deprecated"/>
</module>
<module name="AtclauseOrder">
<property name="target" value="METHOD_DEF, CTOR_DEF, VARIABLE_DEF"/>
<property name="tagOrder" value="@param, @return, @throws, @since, @deprecated, @see"/>
</module>
<!-- Miscellaneous -->
<module name="CommentsIndentation">
<property name="tokens" value="BLOCK_COMMENT_BEGIN" />
</module>
<module name="UpperEll" />
<module name="ArrayTypeStyle" />
<module name="OuterTypeFilename" />
<!-- Modifiers -->
<module name="RedundantModifier" />
<!-- Regexp -->
<module name="RegexpSinglelineJava">
<property name="format" value="^\t* +\t*\S" />
<property name="message"
value="Line has leading space characters; indentation should be performed with tabs only." />
<property name="ignoreComments" value="true" />
</module>
<module name="RegexpSinglelineJava">
<property name="maximum" value="0"/>
<property name="format" value="org\.mockito\..*Mockito\.(when|doThrow|doAnswer)" />
<property name="message"
value="Please use BDDMockito instead of Mockito.(when|doThrow|doAnswer)." />
<property name="ignoreComments" value="true" />
</module>
<module name="RegexpSinglelineJava">
<property name="maximum" value="0"/>
<property name="format" value="org\.junit\.Assert\.assert" />
<property name="message"
value="Please use AssertJ imports." />
<property name="ignoreComments" value="true" />
</module>
<module name="Regexp">
<property name="format" value="[ \t]+$" />
<property name="illegalPattern" value="true" />
<property name="message" value="Trailing whitespace" />
</module>
<!-- Whitespace -->
<module name="GenericWhitespace" />
<module name="MethodParamPad" />
<module name="NoWhitespaceAfter" >
<property name="tokens" value="BNOT, DEC, DOT, INC, LNOT, UNARY_MINUS, UNARY_PLUS, ARRAY_DECLARATOR"/>
</module>
<module name="NoWhitespaceBefore" />
<module name="ParenPad" />
<module name="TypecastParenPad" />
<module name="WhitespaceAfter" />
<module name="WhitespaceAround" />
</module>
</module>

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

@@ -1,11 +1,11 @@
/*
* Copyright 2015-2016 the original author or authors
* Copyright 2015-2016 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
* 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,
@@ -27,6 +27,9 @@ import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAd
import org.springframework.util.StringUtils;
/**
*
* Parser for the message driven channel adapter.
*
* @author Artem Bilan
* @author Gary Russell
*/

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

@@ -19,7 +19,7 @@ package org.springframework.integration.kafka.config.xml;
import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
/**
* The namespace handler for the Kafka namespace
* The namespace handler for the Kafka namespace.
*
* @author Soby Chacko
* @author Gary Russell

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

@@ -27,6 +27,7 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
/**
* Parser for the outbound channel adapter.
*
* @author Soby Chacko
* @author Artem Bilan

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

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -37,10 +37,16 @@ import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* Message-driven channel adapter.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Marius Bogoevici
* @author Gary Russell
*
* TODO: Use the MessagingMessageConverter from spring-kafka
*
*/
public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable {
@@ -137,27 +143,10 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
return getPhase();
}
private class AutoAcknowledgingChannelForwardingMessageListener implements MessageListener<K, V> {
@Override
public void onMessage(ConsumerRecord<K, V> record) {
sendMessage(toMessage(record, null));
}
}
private class AcknowledgingChannelForwardingMessageListener implements AcknowledgingMessageListener<K, V> {
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
sendMessage(toMessage(record, acknowledgment));
}
}
private Message<V> toMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(generateMessageId, generateTimestamp);
KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId,
this.generateTimestamp);
Map<String, Object> rawHeaders = kafkaMessageHeaders.getRawHeaders();
rawHeaders.put(KafkaHeaders.MESSAGE_KEY, record.key());
@@ -180,10 +169,28 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
}
}
@SuppressWarnings("serial")
private static class KafkaMessageHeaders extends MessageHeaders {
private class AutoAcknowledgingChannelForwardingMessageListener implements MessageListener<K, V> {
public KafkaMessageHeaders(boolean generateId, boolean generateTimestamp) {
@Override
public void onMessage(ConsumerRecord<K, V> record) {
sendMessage(toMessage(record, null));
}
}
private class AcknowledgingChannelForwardingMessageListener implements AcknowledgingMessageListener<K, V> {
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
sendMessage(toMessage(record, acknowledgment));
}
}
@SuppressWarnings("serial")
private static final class KafkaMessageHeaders extends MessageHeaders {
private KafkaMessageHeaders(boolean generateId, boolean generateTimestamp) {
super(null, generateId ? null : ID_VALUE_NONE, generateTimestamp ? null : -1L);
}

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

@@ -26,6 +26,11 @@ import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* Kafka Message Handler.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Soby Chacko
* @author Artem Bilan
* @author Gary Russell
@@ -73,6 +78,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
}
/**
* Set the partition expression.
* @param partitionExpression an expression that returns a partition id
* @deprecated as of 1.3, {@link #setPartitionIdExpression(Expression)} should be used instead
*/
@@ -111,12 +117,21 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
? this.messageKeyExpression.getValue(this.evaluationContext, message)
: message.getHeaders().get(KafkaHeaders.MESSAGE_KEY);
// TODO: Add KafkaTemplate method with topic, partition, data only (no key)
if (partitionId == null) {
this.kafkaTemplate.convertAndSend(topic, (K) messageKey, ((V) message.getPayload()));
if (messageKey == null) {
this.kafkaTemplate.send(topic, (V) message.getPayload());
}
else {
this.kafkaTemplate.send(topic, (K) messageKey, (V) message.getPayload());
}
}
else {
this.kafkaTemplate.convertAndSend(topic, partitionId, (K) messageKey, ((V) message.getPayload()));
if (messageKey == null) {
this.kafkaTemplate.send(topic, partitionId, (V) message.getPayload());
}
else {
this.kafkaTemplate.send(topic, partitionId, (K) messageKey, (V) message.getPayload());
}
}
}

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

@@ -1,11 +1,11 @@
/*
* Copyright 2015-2016 the original author or authors
* Copyright 2015-2016 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
* 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,
@@ -16,10 +16,7 @@
package org.springframework.integration.kafka.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -52,15 +49,15 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
@Test
public void testKafkaMessageDrivenChannelAdapterParser() throws Exception {
assertFalse(this.kafkaListener.isAutoStartup());
assertFalse(this.kafkaListener.isRunning());
assertEquals(100, this.kafkaListener.getPhase());
assertSame(this.nullChannel, TestUtils.getPropertyValue(this.kafkaListener, "outputChannel"));
assertSame(this.errorChannel, TestUtils.getPropertyValue(this.kafkaListener, "errorChannel"));
assertThat(this.kafkaListener.isAutoStartup()).isFalse();
assertThat(this.kafkaListener.isRunning()).isFalse();
assertThat(this.kafkaListener.getPhase()).isEqualTo(100);
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "outputChannel")).isSameAs(this.nullChannel);
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "errorChannel")).isSameAs(this.errorChannel);
KafkaMessageListenerContainer<?, ?> container =
TestUtils.getPropertyValue(this.kafkaListener, "messageListenerContainer",
KafkaMessageListenerContainer.class);
assertNotNull(container);
assertThat(container).isNotNull();
}
}

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

@@ -16,8 +16,7 @@
package org.springframework.integration.kafka.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -48,12 +47,12 @@ public class KafkaOutboundAdapterParserTests {
public void testOutboundAdapterConfiguration() {
KafkaProducerMessageHandler<?, ?> messageHandler
= this.appContext.getBean("kafkaOutboundChannelAdapter.handler", KafkaProducerMessageHandler.class);
assertNotNull(messageHandler);
assertEquals(messageHandler.getOrder(), 3);
assertEquals("foo", TestUtils.getPropertyValue(messageHandler, "topicExpression.literalValue"));
assertEquals("'bar'", TestUtils.getPropertyValue(messageHandler, "messageKeyExpression.expression"));
assertEquals("2", TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.expression"));
assertEquals(true, TestUtils.getPropertyValue(messageHandler, "enableHeaderRouting"));
assertThat(messageHandler).isNotNull();
assertThat(messageHandler.getOrder()).isEqualTo(3);
assertThat(TestUtils.getPropertyValue(messageHandler, "topicExpression.literalValue")).isEqualTo("foo");
assertThat(TestUtils.getPropertyValue(messageHandler, "messageKeyExpression.expression")).isEqualTo("'bar'");
assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.expression")).isEqualTo("2");
assertThat(TestUtils.getPropertyValue(messageHandler, "enableHeaderRouting")).isEqualTo(Boolean.TRUE);
}
}

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.kafka.inbound;
import static org.junit.Assert.assertNotNull;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Map;
@@ -66,9 +66,9 @@ public class MessageDrivenAdapterTests {
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic("testTopic1");
template.convertAndSend("foo");
template.send("foo");
Message<?> received = out.receive(10000);
assertNotNull(received);
assertThat(received).isNotNull();
adapter.stop();
}

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

@@ -0,0 +1,141 @@
/*
* Copyright 2016 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.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.springframework.kafka.test.assertj.KafkaConditions.key;
import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
import static org.springframework.kafka.test.assertj.KafkaConditions.value;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
/**
* @author Gary Russell
* @since 2.0
*
*/
public class KafkaProducerMessageHandlerTests {
private static String topic1 = "testTopic1out";
private static String topic2 = "testTopic2out";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2);
private static Consumer<Integer, String> consumer;
@BeforeClass
public static void setUp() throws Exception {
ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(
KafkaTestUtils.consumerProps("testOut", "true", embeddedKafka));
consumer = cf.createConsumer();
final CountDownLatch consumerLatch = new CountDownLatch(1);
consumer.subscribe(Arrays.asList(topic1, topic2), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
consumerLatch.countDown();
}
});
consumer.poll(0); // force assignment
assertThat(consumerLatch.await(30, TimeUnit.SECONDS)).isTrue();
}
@Test
public void testOutbound() {
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
handler.setBeanFactory(mock(BeanFactory.class));
handler.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic1)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.build();
handler.handleMessage(message);
ConsumerRecord<Integer, String> record = getSingleRecord();
assertThat(record).has(key(2));
assertThat(record).has(partition(1));
assertThat(record).has(value("foo"));
message = MessageBuilder.withPayload("bar")
.setHeader(KafkaHeaders.TOPIC, topic1)
.setHeader(KafkaHeaders.PARTITION_ID, 0)
.build();
handler.handleMessage(message);
record = getSingleRecord();
assertThat(record).has(key((Integer) null));
assertThat(record).has(partition(0));
assertThat(record).has(value("bar"));
message = MessageBuilder.withPayload("baz")
.setHeader(KafkaHeaders.TOPIC, topic1)
.build();
handler.handleMessage(message);
record = getSingleRecord();
assertThat(record).has(key((Integer) null));
assertThat(record).has(value("baz"));
}
private ConsumerRecord<Integer, String> getSingleRecord() {
ConsumerRecords<Integer, String> received = getRecords();
assertThat(received.count()).isEqualTo(1);
return received.records(topic1).iterator().next();
}
private ConsumerRecords<Integer, String> getRecords() {
ConsumerRecords<Integer, String> received = consumer.poll(10000);
assertThat(received).isNotNull();
return received;
}
}

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestKey.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2016 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,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
/**

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestPayload.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2016 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,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
/**

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestKey.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2016 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,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
import java.io.Serializable;

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestObject.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2016 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,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
/**

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestPayload.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2016 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,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
import java.io.Serializable;
@@ -27,7 +28,7 @@ public class TestPayload implements Serializable {
private final String part1;
private final String part2;
public TestPayload(final String part1, final String part2){
public TestPayload(final String part1, final String part2) {
this.part1 = part1;
this.part2 = part2;
}