INTEXT-44 - Add Kafka Support
* Add Inbound adapter * Add Outbound adapter * Add serialization * Initial Documentation * Add unit tests * Add async producer * Support Kafka 0.8 * Add Kafka Sample See: https://jira.springsource.org/browse/INTEXT-44
This commit is contained in:
Soby Chacko
committed by
Gunnar Hillert
Gunnar Hillert
parent
0665dc9a86
commit
e46cbfdcb4
@@ -0,0 +1,27 @@
|
||||
/*
|
||||
* Copyright 2002-2013 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.samples.kafka.inbound;
|
||||
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
|
||||
public class InboundRunner {
|
||||
private static final String CONFIG = "kafkaInboundAdapterParserTests-context.xml";
|
||||
|
||||
public static void main(final String args[]) {
|
||||
final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(CONFIG, InboundRunner.class);
|
||||
ctx.start();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
/*
|
||||
* Copyright 2002-2013 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.samples.kafka.outbound;
|
||||
|
||||
import kafka.producer.Partitioner;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
*
|
||||
* This class is for internal use only and therefore is at default access level
|
||||
*/
|
||||
class CustomPartitioner<T> implements Partitioner<T> {
|
||||
/**
|
||||
* Uses the key to calculate a partition bucket id for routing
|
||||
* the data to the appropriate broker partition
|
||||
* @return an integer between 0 and numPartitions-1
|
||||
*/
|
||||
@Override
|
||||
public int partition(final T key, final int numPartitions) {
|
||||
final String s = (String) key;
|
||||
final Integer i = Integer.parseInt(s);
|
||||
return i % numPartitions;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
/*
|
||||
* Copyright 2002-2013 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.samples.kafka.outbound;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
import org.springframework.integration.MessageChannel;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
|
||||
public class OutboundRunner {
|
||||
private static final String CONFIG = "kafkaOutboundAdapterParserTests-context.xml";
|
||||
private static final Log LOG = LogFactory.getLog(OutboundRunner.class);
|
||||
|
||||
public static void main(final String args[]) {
|
||||
final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(CONFIG, OutboundRunner.class);
|
||||
ctx.start();
|
||||
|
||||
final MessageChannel channel = ctx.getBean("inputToKafka", MessageChannel.class);
|
||||
LOG.info(channel.getClass());
|
||||
|
||||
//sending 100,000 messages to Kafka server for topic test1
|
||||
for (int i = 0; i < 500; i++) {
|
||||
channel.send(
|
||||
MessageBuilder.withPayload("hello Fom ob adapter test1 - " + i)
|
||||
.setHeader("messageKey", String.valueOf(i))
|
||||
.setHeader("topic", "test1").build());
|
||||
|
||||
LOG.info("message sent " + i);
|
||||
}
|
||||
|
||||
//sending 5,000 messages to kafka server for topic test2
|
||||
for (int i = 0; i < 50; i++) {
|
||||
channel.send(
|
||||
MessageBuilder.withPayload("hello Fom ob adapter test2 - " + i)
|
||||
.setHeader("messageKey", String.valueOf(i))
|
||||
.setHeader("topic", "test2").build());
|
||||
|
||||
LOG.info("message sent " + i);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
/*
|
||||
* Copyright 2002-2013 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.samples.kafka.outbound;
|
||||
|
||||
import org.springframework.integration.Message;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.transformer.Transformer;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
*/
|
||||
public class PartitionlessTransformer implements Transformer {
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public Message<?> transform(final Message<?> message) {
|
||||
|
||||
final Map<String, Map<Integer, List<Object>>> origData =
|
||||
(Map<String, Map<Integer, List<Object>>>) message.getPayload();
|
||||
|
||||
final Map<String, List<Object>> nonPartitionedData = new HashMap<String, List<Object>>();
|
||||
|
||||
for(final String topic : origData.keySet()) {
|
||||
final Map<Integer, List<Object>> partitionedData = origData.get(topic);
|
||||
final Collection<List<Object>> nonPartitionedDataFromTopic = partitionedData.values();
|
||||
|
||||
final List<Object> mergedList = new ArrayList<Object>();
|
||||
|
||||
for (final List<Object> l : nonPartitionedDataFromTopic){
|
||||
mergedList.addAll(l);
|
||||
}
|
||||
|
||||
nonPartitionedData.put(topic, mergedList);
|
||||
}
|
||||
|
||||
return MessageBuilder.withPayload(nonPartitionedData).build();
|
||||
}
|
||||
}
|
||||
10
samples/kafka/src/main/resources/log4j.properties
Normal file
10
samples/kafka/src/main/resources/log4j.properties
Normal file
@@ -0,0 +1,10 @@
|
||||
log4j.rootCategory=WARN, stdout
|
||||
|
||||
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
|
||||
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
|
||||
log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n
|
||||
|
||||
#log4j.category.org.springframework.integration=WARN
|
||||
log4j.category.org.springframework.integration.kafka=INFO
|
||||
log4j.category.org.springframework.integration.samples.kafka=INFO
|
||||
log4j.category.kafka.consumer=ERROR
|
||||
@@ -0,0 +1,56 @@
|
||||
<?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:stream="http://www.springframework.org/schema/integration/stream"
|
||||
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
|
||||
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">
|
||||
|
||||
<int:channel id="inputFromKafka"/>
|
||||
|
||||
<int:channel id="inputFromTransformer"/>
|
||||
|
||||
<stream:stdout-channel-adapter id="stdout" channel="inputFromKafka" append-newline="true"/>
|
||||
|
||||
<int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181"
|
||||
zk-connection-timeout="6000"
|
||||
zk-session-timeout="6000"
|
||||
zk-sync-time="2000"/>
|
||||
|
||||
<int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter"
|
||||
kafka-consumer-context-ref="consumerContext"
|
||||
auto-startup="false"
|
||||
channel="inputFromKafka">
|
||||
<int:poller fixed-delay="1" time-unit="MILLISECONDS"/>
|
||||
</int-kafka:inbound-channel-adapter>
|
||||
|
||||
<bean id="kafkaDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaDecoder">
|
||||
<constructor-arg type="java.lang.Class" value="java.lang.String"/>
|
||||
</bean>
|
||||
|
||||
<int-kafka:consumer-context id="consumerContext"
|
||||
consumer-timeout="1000"
|
||||
zookeeper-connect="zookeeperConnect">
|
||||
<int-kafka:consumer-configurations>
|
||||
<int-kafka:consumer-configuration group-id="default"
|
||||
value-decoder="kafkaDecoder"
|
||||
key-decoder="kafkaDecoder"
|
||||
max-messages="5000">
|
||||
<int-kafka:topic id="test1" streams="4"/>
|
||||
</int-kafka:consumer-configuration>
|
||||
<int-kafka:consumer-configuration group-id="default1"
|
||||
max-messages="5">
|
||||
<int-kafka:topic id="test2" streams="4"/>
|
||||
</int-kafka:consumer-configuration>
|
||||
</int-kafka:consumer-configurations>
|
||||
</int-kafka:consumer-context>
|
||||
|
||||
<bean id="partitionlessTransformer" class="org.springframework.integration.samples.kafka.outbound.PartitionlessTransformer"/>
|
||||
|
||||
<int:transformer ref="partitionlessTransformer" method="transform"
|
||||
input-channel="inputFromKafka"
|
||||
output-channel="inputFromTransformer"/>
|
||||
</beans>
|
||||
@@ -0,0 +1,45 @@
|
||||
<?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
|
||||
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-kafka:outbound-channel-adapter kafka-producer-context-ref="kafkaProducerContext"
|
||||
auto-startup="false"
|
||||
channel="inputToKafka">
|
||||
<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"/>
|
||||
|
||||
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroBackedKafkaEncoder">
|
||||
<constructor-arg value="java.lang.String" />
|
||||
</bean>
|
||||
|
||||
<bean id="customPartitioner" class="org.springframework.integration.samples.kafka.outbound.CustomPartitioner"/>
|
||||
|
||||
<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"
|
||||
partitioner="customPartitioner"/>
|
||||
<int-kafka:producer-configuration broker-list="localhost:9092"
|
||||
topic="test2"
|
||||
compression-codec="default"/>
|
||||
</int-kafka:producer-configurations>
|
||||
</int-kafka:producer-context>
|
||||
</beans>
|
||||
Reference in New Issue
Block a user