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

INTEXT-121 Add SmartLifecycle to ProducerContext

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

When using `async` producers, the buffered messages need to be flushed.

Previously, when the context was stopped, such messages were lost.

- Implement `SmartLifecyle` in the producer context and propagate the `stop()` to the underlying producer(s).
- Remove unused attributes from the consumer parser.
- Add a test case to show the buffered message is received after closing the producer.
- Add KafkaRunning JUnit `@Rule`

Add KafkaRunning Rule to Parser Tests

Polishing
This commit is contained in:
Gary Russell
2014-12-02 21:50:37 +02:00
committed by Artem Bilan
parent 503ce78395
commit 0e2808c014
15 changed files with 459 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java
View File

@@ -45,6 +45,7 @@ import org.springframework.util.xml.DomUtils;
* @author Rajasekar Elango
* @author Artem Bilan
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @since 0.5
*/
public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionParser {
@@ -78,10 +79,6 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars
"value-decoder");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(consumerMetadataBuilder, consumerConfiguration,
"key-decoder");
IntegrationNamespaceUtils.setValueIfAttributeDefined(consumerMetadataBuilder, consumerConfiguration,
"key-class-type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(consumerMetadataBuilder, consumerConfiguration,
"value-class-type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(consumerConfigurationBuilder, consumerConfiguration,
"max-messages");
IntegrationNamespaceUtils.setValueIfAttributeDefined(consumerMetadataBuilder, parentElem,

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

@@ -37,6 +37,7 @@ import org.springframework.util.xml.DomUtils;
/**
* @author Soby Chacko
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @since 0.5
*/
public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionParser {
@@ -50,17 +51,24 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars
protected void doParse(final Element element, final ParserContext parserContext, final BeanDefinitionBuilder builder) {
super.doParse(element, parserContext, builder);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
final Element topics = DomUtils.getChildElementByTagName(element, "producer-configurations");
parseProducerConfigurations(topics, parserContext, builder, element);
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !"producer-properties".equals(attributeName) && super.isEligibleAttribute(attributeName);
}
private void parseProducerConfigurations(Element topics, ParserContext parserContext,
BeanDefinitionBuilder builder, Element parentElem) {
Map<String, BeanMetadataElement> producerConfigurationsMap = new ManagedMap<String, BeanMetadataElement>();
for (Element producerConfiguration : DomUtils.getChildElementsByTagName(topics, "producer-configuration")) {
BeanDefinitionBuilder producerConfigurationBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ProducerConfiguration.class);
BeanDefinitionBuilder producerMetadataBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ProducerMetadata.class);
@@ -100,11 +108,11 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars
AbstractBeanDefinition producerFactoryBeanDefinition = producerFactoryBuilder.getBeanDefinition();
producerConfigurationBuilder.addConstructorArgValue(producerMetadataBeanDefinition);
producerConfigurationBuilder.addConstructorArgValue(producerFactoryBeanDefinition);
AbstractBeanDefinition producerConfigurationBeanDefinition =
producerConfigurationBuilder.getBeanDefinition();
BeanDefinitionBuilder.genericBeanDefinition(ProducerConfiguration.class)
.addConstructorArgValue(producerMetadataBeanDefinition)
.addConstructorArgValue(producerFactoryBeanDefinition)
.getBeanDefinition();
producerConfigurationsMap.put(producerConfiguration.getAttribute("topic"),
producerConfigurationBeanDefinition);
}

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.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.
@@ -21,6 +21,7 @@ import org.springframework.messaging.Message;
/**
* @author Soby Chacko
* @author Gary Russell
* @since 0.5
*/
public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler {
@@ -39,4 +40,10 @@ public class KafkaProducerMessageHandler<K,V> extends AbstractMessageHandler {
protected void handleMessageInternal(final Message<?> message) throws Exception {
kafkaProducerContext.send(message);
}
@Override
public String getComponentType() {
return "kafka:outbound-channel-adapter";
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 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.
@@ -19,21 +19,28 @@ 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;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.Message;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @since 0.5
*/
public class KafkaProducerContext<K, V> {
public class KafkaProducerContext<K, V> implements SmartLifecycle, NamedComponent, BeanNameAware {
private static final Log LOGGER = LogFactory.getLog(KafkaProducerContext.class);
private static final Log logger = LogFactory.getLog(KafkaProducerContext.class);
private final AtomicBoolean running = new AtomicBoolean();
private volatile Map<String, ProducerConfiguration<K, V>> producerConfigurations;
@@ -41,23 +48,11 @@ public class KafkaProducerContext<K, V> {
private Properties producerProperties;
public void send(final Message<?> message) throws Exception {
if (message.getHeaders().containsKey("topic")) {
ProducerConfiguration<K, V> producerConfiguration =
getTopicConfiguration(message.getHeaders().get("topic", String.class));
if (producerConfiguration != null) {
producerConfiguration.send(message);
}
}
// if there is a single producer configuration then use that config to send message.
else if (this.theProducerConfiguration != null) {
this.theProducerConfiguration.send(message);
}
else {
throw new IllegalStateException("Could not send messages as there are multiple producer configurations " +
"with no topic information found from the message header.");
}
}
private String beanName = "not_specified";
private int phase = 0;
private boolean autoStartup = true;
public ProducerConfiguration<K, V> getTopicConfiguration(final String topic) {
if (this.theProducerConfiguration != null) {
@@ -73,7 +68,7 @@ public class KafkaProducerContext<K, V> {
return producerConfiguration;
}
}
LOGGER.error("No producer-configuration defined for topic " + topic + ". Cannot send message");
logger.error("No producer-configuration defined for topic " + topic + ". Cannot send message");
return null;
}
@@ -97,10 +92,133 @@ public class KafkaProducerContext<K, V> {
}
/**
* @return Returns the producerProperties.
* @return the producerProperties.
*/
public Properties getProducerProperties() {
return this.producerProperties;
}
/**
* @return the component type.
* @since 1.0
*/
@Override
public String getComponentType() {
return "kafka:producer-context";
}
/**
* @param name the bean name.
* @since 1.0
*/
@Override
public void setBeanName(String name) {
this.beanName = name;
}
/**
* @param phase the phase to set.
* @see SmartLifecycle
* @since 1.0
*/
public void setPhase(int phase) {
this.phase = phase;
}
/**
* @param autoStartup the autoStartup to set.
* @see SmartLifecycle
* @since 1.0
*/
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
/**
* @return the component name.
* @since 1.0
*/
@Override
public String getComponentName() {
return this.beanName;
}
protected void doStart() {
}
protected void doStop() {
if (this.producerConfigurations != null) {
for (ProducerConfiguration<?, ?> producerConfiguration : this.producerConfigurations.values()) {
producerConfiguration.stop();
}
}
}
@Override
public final void start() {
if (this.running.compareAndSet(false, true)) {
doStart();
}
else {
if (logger.isDebugEnabled()) {
logger.debug(getComponentType() + ":" + getComponentName() + " is already running");
}
}
}
@Override
public final void stop() {
if (this.running.compareAndSet(true, false)) {
doStop();
}
else {
if (logger.isDebugEnabled()) {
logger.debug(getComponentType() + ":" + getComponentName() + " is not running");
}
}
}
@Override
public boolean isRunning() {
return this.running.get();
}
@Override
public int getPhase() {
return this.phase;
}
@Override
public boolean isAutoStartup() {
return this.autoStartup;
}
@Override
public void stop(Runnable callback) {
stop();
callback.run();
}
public void send(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);
}
}
// if there is a single producer configuration then use that config to send message.
else if (this.theProducerConfiguration != null) {
this.theProducerConfiguration.send(message);
}
else {
throw new IllegalStateException("Could not send messages as there are multiple producer configurations " +
"with no topic information found from the message header.");
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 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.
@@ -25,6 +25,7 @@ import org.apache.commons.lang.builder.HashCodeBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.util.Assert;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
@@ -34,6 +35,7 @@ import kafka.serializer.DefaultEncoder;
* @author Soby Chacko
* @author Rajasekar Elango
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @since 0.5
*/
public class ProducerConfiguration<K, V> {
@@ -42,7 +44,9 @@ public class ProducerConfiguration<K, V> {
private final ProducerMetadata<K, V> producerMetadata;
public ProducerConfiguration(final ProducerMetadata<K, V> producerMetadata, final Producer<K, V> producer) {
public ProducerConfiguration(ProducerMetadata<K, V> producerMetadata, Producer<K, V> producer) {
Assert.notNull(producerMetadata);
Assert.notNull(producer);
this.producerMetadata = producerMetadata;
this.producer = producer;
}
@@ -120,4 +124,8 @@ public class ProducerConfiguration<K, V> {
return "ProducerConfiguration [producerMetadata=" + this.producerMetadata + "]";
}
public void stop() {
this.producer.close();
}
}

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

@@ -214,6 +214,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup" />
</xsd:complexType>
</xsd:element>

10
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 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.
@@ -19,15 +19,17 @@ package org.springframework.integration.kafka.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import kafka.consumer.Blacklist;
import org.hamcrest.Matchers;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.ConsumerConfiguration;
import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
@@ -38,12 +40,16 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Soby Chacko
* @author Artem Bilan
* @author Gary Russell
* @since 0.5
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaConsumerContextParserTests<K, V> {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@Autowired
private ApplicationContext appContext;

9
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundAdapterParserTests.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.
@@ -16,22 +16,29 @@
package org.springframework.integration.kafka.config.xml;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Soby Chacko
* @author Gary Russell
* @since 0.5
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaInboundAdapterParserTests {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@Autowired
private ApplicationContext appContext;

7
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMultiConsumerContextParserTests.java
View File

@@ -16,10 +16,13 @@
package org.springframework.integration.kafka.config.xml;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.ConsumerConfiguration;
import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
@@ -28,11 +31,15 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Ilayaperumal Gopinathan
* @author Gary Russell
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaMultiConsumerContextParserTests<K,V> {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@Autowired
private ApplicationContext appContext;

8
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.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.
@@ -16,6 +16,7 @@
package org.springframework.integration.kafka.config.xml;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -23,18 +24,23 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
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.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Soby Chacko
* @author Gary Russell
* @since 0.5
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaOutboundAdapterParserTests<K, V> {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@Autowired
private ApplicationContext appContext;

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

@@ -27,7 +27,8 @@
<context:property-placeholder properties-ref="placeholderProperties"/>
<int-kafka:producer-context id="producerContext" producer-properties="producerProperties">
<int-kafka:producer-context id="producerContext" producer-properties="producerProperties"
auto-startup="false" phase="123">
<int-kafka:producer-configurations>
<int-kafka:producer-configuration broker-list="${brokerList1}"
key-class-type="java.lang.String"

41
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParserTests.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,29 +15,41 @@
*/
package org.springframework.integration.kafka.config.xml;
import org.junit.Assert;
import kafka.javaapi.producer.Producer;
import kafka.serializer.Encoder;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import java.util.Map;
import kafka.javaapi.producer.Producer;
import kafka.serializer.Encoder;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
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.ProducerMetadata;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Soby Chacko
* @author Gary Russell
* @since 0.5
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaProducerContextParserTests<K,V,T> {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@Autowired
private ApplicationContext appContext;
@@ -73,5 +85,8 @@ public class KafkaProducerContextParserTests<K,V,T> {
final Producer<K,V> producerTest2 = producerConfigurationTest2.getProducer();
Assert.assertEquals(producerConfigurationTest2, new ProducerConfiguration<K,V>(producerMetadataTest2, producerTest2));
assertFalse(TestUtils.getPropertyValue(producerContext, "autoStartup", Boolean.class));
assertEquals(123, TestUtils.getPropertyValue(producerContext, "phase"));
}
}

9
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ZookeeperConnectParserTests.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.
@@ -16,23 +16,30 @@
package org.springframework.integration.kafka.config.xml;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.kafka.core.ZookeeperConnectDefaults;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Soby Chacko
* @author Gary Russell
* @since 0.5
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class ZookeeperConnectParserTests {
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@Autowired
private ApplicationContext appContext;

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

@@ -0,0 +1,122 @@
/*
* 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.outbound;
import static org.junit.Assert.assertNotNull;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.serializer.common.StringDecoder;
import org.springframework.integration.kafka.serializer.common.StringEncoder;
import org.springframework.integration.kafka.support.ConsumerConfigFactoryBean;
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.KafkaProducerContext;
import org.springframework.integration.kafka.support.MessageLeftOverTracker;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerFactoryBean;
import org.springframework.integration.kafka.support.ProducerMetadata;
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
*
*/
public class OutboundTests {
private static final String TOPIC = "springintegrationtest";
@ClassRule
public static KafkaRunning kafkaRunning = KafkaRunning.isRunning();
@Test
public void testAsyncProducerFlushed() throws Exception {
KafkaConsumerContext<String, String> consumerContext = createConsumer();
// pre-consume to start the receiver because the high-level API doesn't support --from-beginning
consumerContext.receive();
KafkaProducerContext<String, String> kafkaProducerContext = new KafkaProducerContext<String, String>();
ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>(TOPIC);
producerMetadata.setValueClassType(String.class);
producerMetadata.setKeyClassType(String.class);
Encoder<String> encoder = new StringEncoder<String>();
producerMetadata.setValueEncoder(encoder);
producerMetadata.setKeyEncoder(encoder);
producerMetadata.setAsync(true);
Properties props = new Properties();
props.put("queue.buffering.max.ms", "15000");
ProducerFactoryBean<String, String> producer =
new ProducerFactoryBean<String, String>(producerMetadata, "localhost:9092", props);
ProducerConfiguration<String, String> config =
new ProducerConfiguration<String, String>(producerMetadata, producer.getObject());
kafkaProducerContext.setProducerConfigurations(Collections.singletonMap(TOPIC, config));
KafkaProducerMessageHandler<String, String> handler = new KafkaProducerMessageHandler<String, String>(kafkaProducerContext);
handler.handleMessage(MessageBuilder.withPayload("foo")
.setHeader("messagekey", "3")
.setHeader("topic", TOPIC)
.build());
kafkaProducerContext.stop();
Message<Map<String, Map<Integer, List<Object>>>> received = consumerContext.receive();
assertNotNull(received);
consumerContext.destroy();
}
private KafkaConsumerContext<String, String> createConsumer() throws Exception {
KafkaConsumerContext<String, String> consumerContext = new KafkaConsumerContext<String, String>();
ZookeeperConnect zookeeperConnect = new ZookeeperConnect();
zookeeperConnect.setZkConnect("localhost:2181");
consumerContext.setZookeeperConnect(zookeeperConnect);
ConsumerMetadata<String, String> consumerMetadata = new ConsumerMetadata<String, String>();
consumerMetadata.setGroupId("foo");
Decoder<String> decoder = new StringDecoder();
consumerMetadata.setValueDecoder(decoder);
consumerMetadata.setKeyDecoder(decoder);
consumerMetadata.setTopicStreamMap(Collections.singletonMap(TOPIC, 1));
Properties consumerProps = new Properties();
consumerProps.put("consumer.timeout.ms", "500");
ConsumerConfigFactoryBean<String, String> consumerConfigFactoryBean = new ConsumerConfigFactoryBean<String, String>(
consumerMetadata, zookeeperConnect, consumerProps);
ConsumerConfig consumerConfig = consumerConfigFactoryBean.getObject();
ConsumerConnectionProvider consumerConnectionProvider = new ConsumerConnectionProvider(consumerConfig);
MessageLeftOverTracker<String, String> messageLeftOverTracker = new MessageLeftOverTracker<String, String>();
ConsumerConfiguration<String, String> cConfig = new ConsumerConfiguration<String, String>(consumerMetadata,
consumerConnectionProvider, messageLeftOverTracker);
cConfig.setMaxMessages(1);
consumerContext.setConsumerConfigurations(Collections.singletonMap("foo", cConfig));
return consumerContext;
}
}

95
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java Normal file
View File

@@ -0,0 +1,95 @@
/*
* 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.rule;
import java.io.IOException;
import java.net.Socket;
import javax.net.SocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Assume;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
/**
* <p>
* A rule that prevents integration tests from failing if the Kafka server is not running or not
* accessible. If the Kafka server is not running in the background all the tests here will simply be skipped because
* of a violated assumption (showing as successful).
* <p>
* The rule can be declared as static so that it only has to check once for all tests in the enclosing test case, but
* there isn't a lot of overhead in making it non-static.
*
* @author Dave Syer
* @author Artem Bilan
* @author Gary Russell
*
* @since 1.0
*/
public class KafkaRunning extends TestWatcher {
public static final int KAFKA_PORT = 9092;
public static final int ZOOKEEPER_PORT = 2181;
private static final Log logger = LogFactory.getLog(KafkaRunning.class);
/**
* @return a new rule that assumes an existing running broker
*/
public static KafkaRunning isRunning() {
return new KafkaRunning();
}
@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();
}
catch (final 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);
}
}