diff --git a/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java b/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java index e5f1e226..fccd7350 100644 --- a/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java +++ b/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java @@ -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 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 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 producerFactory() { + Map 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 producerMetadata = new ProducerMetadata<>(this.topic, String.class, - String.class, new StringSerializer(), new StringSerializer()); - Properties props = new Properties(); - props.put("linger.ms", "1000"); - ProducerFactoryBean producer = - new ProducerFactoryBean<>(producerMetadata, this.brokerAddress, props); - ProducerConfiguration config = - new ProducerConfiguration<>(producerMetadata, producer.getObject()); - Map> producerConfigurationMap = - Collections.>singletonMap(this.topic, config); - kafkaProducerContext.setProducerConfigurations(producerConfigurationMap); - return kafkaProducerContext; + public KafkaMessageListenerContainer 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 consumerFactory() { + Map 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 + adapter(KafkaMessageListenerContainer container) { + KafkaMessageDrivenChannelAdapter 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; } diff --git a/basic/kafka/src/test/java/org/springframework/integration/samples/kafka/ApplicationTests.java b/basic/kafka/src/test/java/org/springframework/integration/samples/kafka/ApplicationTests.java deleted file mode 100644 index 9aa22b5e..00000000 --- a/basic/kafka/src/test/java/org/springframework/integration/samples/kafka/ApplicationTests.java +++ /dev/null @@ -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() { - } - -} diff --git a/basic/kafka/src/test/java/org/springframework/integration/samples/kafka/KafkaRunning.java b/basic/kafka/src/test/java/org/springframework/integration/samples/kafka/KafkaRunning.java deleted file mode 100644 index 1153900d..00000000 --- a/basic/kafka/src/test/java/org/springframework/integration/samples/kafka/KafkaRunning.java +++ /dev/null @@ -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 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); - } - -} diff --git a/build.gradle b/build.gradle index 65108b7a..9bc1b106 100644 --- a/build.gradle +++ b/build.gradle @@ -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' }