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

Kafka - Update to S-I-K 2.0

Browse Source 
Polishing
This commit is contained in:
Gary Russell
2016-03-11 15:54:01 -05:00
committed by Artem Bilan
parent 689279656e
commit f943f5e20f
4 changed files with 62 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java
View File

@@ -16,10 +16,15 @@
package org.springframework.integration.samples.kafka;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.I0Itec.zkclient.ZkClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
@@ -31,29 +36,25 @@ import org.springframework.context.annotation.Bean;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.kafka.core.BrokerAddress;
import org.springframework.integration.kafka.core.BrokerAddressListConfiguration;
import org.springframework.integration.kafka.core.ConnectionFactory;
import org.springframework.integration.kafka.core.DefaultConnectionFactory;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.integration.kafka.listener.KafkaTopicOffsetManager;
import org.springframework.integration.kafka.listener.OffsetManager;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.integration.kafka.serializer.common.StringDecoder;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerFactoryBean;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.integration.kafka.util.TopicUtils;
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.listener.KafkaMessageListenerContainer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import kafka.admin.AdminUtils;
import kafka.common.TopicExistsException;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
/**
* @author Gary Russell
* @since 4.2
@@ -95,64 +96,54 @@ public class Application {
@ServiceActivator(inputChannel = "toKafka")
@Bean
public MessageHandler handler() throws Exception {
KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(producerContext());
KafkaProducerMessageHandler<String, String> handler =
new KafkaProducerMessageHandler<>(kafkaTemplate());
handler.setTopicExpression(new LiteralExpression(this.topic));
handler.setMessageKeyExpression(new LiteralExpression(this.messageKey));
return handler;
}
@Bean
public ConnectionFactory kafkaBrokerConnectionFactory() throws Exception {
return new DefaultConnectionFactory(kafkaConfiguration());
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public org.springframework.integration.kafka.core.Configuration kafkaConfiguration() {
BrokerAddressListConfiguration configuration = new BrokerAddressListConfiguration(
BrokerAddress.fromAddress(this.brokerAddress));
configuration.setSocketTimeout(500);
return configuration;
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public KafkaProducerContext producerContext() throws Exception {
KafkaProducerContext kafkaProducerContext = new KafkaProducerContext();
ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<>(this.topic, String.class,
String.class, new StringSerializer(), new StringSerializer());
Properties props = new Properties();
props.put("linger.ms", "1000");
ProducerFactoryBean<String, String> producer =
new ProducerFactoryBean<>(producerMetadata, this.brokerAddress, props);
ProducerConfiguration<String, String> config =
new ProducerConfiguration<>(producerMetadata, producer.getObject());
Map<String, ProducerConfiguration<?, ?>> producerConfigurationMap =
Collections.<String, ProducerConfiguration<?, ?>>singletonMap(this.topic, config);
kafkaProducerContext.setProducerConfigurations(producerConfigurationMap);
return kafkaProducerContext;
public KafkaMessageListenerContainer<String, String> container() throws Exception {
return new KafkaMessageListenerContainer<>(consumerFactory(), new TopicPartition(this.topic, 0));
}
@Bean
public OffsetManager offsetManager() {
return new KafkaTopicOffsetManager(new ZookeeperConnect(this.zookeeperConnect), "si-offsets");
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "siTestGroup");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean
public KafkaMessageListenerContainer container(OffsetManager offsetManager) throws Exception {
final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(
kafkaBrokerConnectionFactory(), new Partition(this.topic, 0));
kafkaMessageListenerContainer.setOffsetManager(offsetManager);
kafkaMessageListenerContainer.setMaxFetch(100);
kafkaMessageListenerContainer.setConcurrency(1);
return kafkaMessageListenerContainer;
}
@Bean
public KafkaMessageDrivenChannelAdapter adapter(KafkaMessageListenerContainer container) {
KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter(container);
StringDecoder decoder = new StringDecoder();
kafkaMessageDrivenChannelAdapter.setKeyDecoder(decoder);
kafkaMessageDrivenChannelAdapter.setPayloadDecoder(decoder);
public KafkaMessageDrivenChannelAdapter<String, String>
adapter(KafkaMessageListenerContainer<String, String> container) {
KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter<>(container);
kafkaMessageDrivenChannelAdapter.setOutputChannel(received());
return kafkaMessageDrivenChannelAdapter;
}
@@ -182,7 +173,14 @@ public class Application {
@Override
public void start() {
TopicUtils.ensureTopicCreated(this.zkConnect, this.topic, 1, 1);
ZkUtils zkUtils = new ZkUtils(new ZkClient(this.zkConnect, 6000, 6000,
ZKStringSerializer$.MODULE$), null, false);
try {
AdminUtils.createTopic(zkUtils, topic, 1, 1, new Properties());
}
catch (TopicExistsException e) {
// no-op
}
this.running = true;
}

41
basic/kafka/src/test/java/org/springframework/integration/samples/kafka/ApplicationTests.java
View File

@@ -1,41 +0,0 @@
package org.springframework.integration.samples.kafka;
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @author Artem Bilan
* @since 4.2
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = Application.class)
public class ApplicationTests {
@ClassRule
public static KafkaRunning kafkaRunning = new KafkaRunning();
@Test
public void contextLoads() {
}
}

78
basic/kafka/src/test/java/org/springframework/integration/samples/kafka/KafkaRunning.java
View File

@@ -1,78 +0,0 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.samples.kafka;
import org.I0Itec.zkclient.ZkClient;
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;
import org.springframework.integration.kafka.core.BrokerAddress;
import org.springframework.integration.kafka.core.ZookeeperConnectDefaults;
import com.gs.collections.impl.utility.ListIterate;
import kafka.cluster.Broker;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import scala.collection.JavaConversions;
import scala.collection.Seq;
/**
* * 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).
* 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
* @author Marius Bogoevici
* @since 4.2
*/
public class KafkaRunning extends TestWatcher {
private static final String ZOOKEEPER_CONNECT_STRING = ZookeeperConnectDefaults.ZK_CONNECT;
private static final Log logger = LogFactory.getLog(KafkaRunning.class);
@Override
public Statement apply(Statement base, Description description) {
try {
ZkClient zkClient = new ZkClient(ZOOKEEPER_CONNECT_STRING, 1000, 1000, ZKStringSerializer$.MODULE$);
Seq<Broker> allBrokersInCluster = ZkUtils.getAllBrokersInCluster(zkClient);
BrokerAddress[] brokerAddresses = ListIterate
.collect(JavaConversions.asJavaList(allBrokersInCluster),
broker -> new BrokerAddress(broker.host(), broker.port()))
.toArray(new BrokerAddress[allBrokersInCluster.size()]);
if (brokerAddresses.length == 0) {
throw new IllegalStateException("No running Kafka brokers");
}
}
catch (Exception e) {
logger.warn("Not executing tests because basic connectivity test failed");
Assume.assumeNoException(e);
}
return super.apply(base, description);
}
}

10
build.gradle
View File

@@ -26,7 +26,7 @@ allprojects {
group = 'org.springframework.integration.samples'
repositories {
// mavenLocal()
mavenLocal()
maven { url 'http://repo.spring.io/libs-snapshot' }
maven { url 'http://repo.spring.io/libs-milestone' }
maven { url 'http://repo.spring.io/libs-staging-local' }
@@ -196,8 +196,9 @@ subprojects { subproject ->
slf4jVersion = '1.7.11'
springIntegrationVersion = '4.3.0.M1'
springIntegrationDslVersion = '1.1.2.RELEASE'
springIntegrationKafkaVersion = '1.3.0.RELEASE'
springIntegrationKafkaVersion = '2.0.0.BUILD-SNAPSHOT'
springIntegrationSplunkVersion = '1.1.0.RELEASE'
springKafkaVersion = '1.0.0.M1'
springVersion = '4.2.5.RELEASE'
springSecurityVersion = '4.0.2.RELEASE'
springWebFlowVersion = '2.3.3.RELEASE'
@@ -600,10 +601,13 @@ project('kafka') {
dependencies {
compile 'org.springframework.boot:spring-boot-starter-integration'
compile("org.springframework.integration:spring-integration-kafka:$springIntegrationKafkaVersion") {
compile ("org.springframework.integration:spring-integration-kafka:$springIntegrationKafkaVersion") {
exclude group: 'org.slf4j'
}
compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion"
compile ("org.springframework.kafka:spring-kafka-test:$springKafkaVersion") {
exclude group: 'org.slf4j'
}
testCompile 'org.springframework.boot:spring-boot-starter-test'
}