SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade binder to Kafka 0.9

Browse Source 
- Upgrade Spring Integration Kafka from 1.3 to 2.x
- Use Spring Kafka 1.x
- Remove wondowing configuration
- Use the new Spring Kafka and Spring Integration Kafka APIs
This commit is contained in:
Soby Chacko
2016-06-22 18:02:57 -04:00
committed by Marius Bogoevici
parent 9cf07a3df4
commit 1bf14ac390
16 changed files with 1049 additions and 991 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pom.xml
View File

@@ -7,10 +7,11 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build</artifactId>
<version>1.1.1.RELEASE</version>
<version>1.1.2.BUILD-SNAPSHOT</version>
<relativePath />
</parent>
<properties>
<java.version>1.7</java.version>
<spring-boot.version>1.4.0.BUILD-SNAPSHOT</spring-boot.version>
</properties>
<dependencyManagement>

2
spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides
View File

@@ -1 +1 @@
provides: spring-cloud-stream-binder-kafka
provides: spring-cloud-starter-stream-kafka

52
spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java
View File

@@ -86,12 +86,12 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
}
public KafkaServer getKafkaServer() {
return kafkaServer;
return this.kafkaServer;
}
public String getZkConnectString() {
if (embedded) {
return zookeeper.getConnectString();
if (this.embedded) {
return this.zookeeper.getConnectString();
}
else {
return DEFAULT_ZOOKEEPER_CONNECT;
@@ -103,8 +103,8 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
}
public String getBrokerAddress() {
if (embedded) {
return kafkaServer.config().hostName() + ":" + kafkaServer.config().port();
if (this.embedded) {
return this.kafkaServer.config().hostName() + ":" + this.kafkaServer.config().port();
}
else {
return DEFAULT_KAFKA_CONNECT;
@@ -114,35 +114,35 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
@Override
protected void obtainResource() throws Exception {
if (!hasFailedAlready) {
if (embedded) {
if (this.embedded) {
try {
log.debug("Starting Zookeeper");
zookeeper = new EmbeddedZookeeper("127.0.0.1:" + SocketUtils.findAvailableTcpPort());
log.debug("Started Zookeeper at " + zookeeper.getConnectString());
this.zookeeper = new EmbeddedZookeeper("127.0.0.1:" + SocketUtils.findAvailableTcpPort());
log.debug("Started Zookeeper at " + this.zookeeper.getConnectString());
try {
int zkConnectionTimeout = 10000;
int zkSessionTimeout = 10000;
zkClient = new ZkClient(getZkConnectString(), zkSessionTimeout, zkConnectionTimeout,
this.zkClient = new ZkClient(getZkConnectString(), zkSessionTimeout, zkConnectionTimeout,
ZKStringSerializer$.MODULE$);
}
catch (Exception e) {
zookeeper.shutdown();
this.zookeeper.shutdown();
throw e;
}
try {
log.debug("Creating Kafka server");
Properties brokerConfigProperties = brokerConfig;
brokerConfig.put("zookeeper.connect", zookeeper.getConnectString());
brokerConfig.put("auto.create.topics.enable", "false");
brokerConfig.put("delete.topic.enable", "true");
kafkaServer = TestUtils.createServer(new KafkaConfig(brokerConfigProperties),
Properties brokerConfigProperties = this.brokerConfig;
this.brokerConfig.put("zookeeper.connect", this.zookeeper.getConnectString());
this.brokerConfig.put("auto.create.topics.enable", "false");
this.brokerConfig.put("delete.topic.enable", "true");
this.kafkaServer = TestUtils.createServer(new KafkaConfig(brokerConfigProperties),
SystemTime$.MODULE$);
log.debug("Created Kafka server at " + kafkaServer.config().hostName() + ":"
+ kafkaServer.config().port());
log.debug("Created Kafka server at " + this.kafkaServer.config().hostName() + ":"
+ this.kafkaServer.config().port());
}
catch (Exception e) {
zookeeper.shutdown();
zkClient.close();
this.zookeeper.shutdown();
this.zkClient.close();
throw e;
}
}
@@ -153,7 +153,7 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
}
else {
this.zkClient = new ZkClient(DEFAULT_ZOOKEEPER_CONNECT, 10000, 10000, ZKStringSerializer$.MODULE$);
if (ZkUtils.getAllBrokersInCluster(zkClient).size() == 0) {
if (ZkUtils.getAllBrokersInCluster(this.zkClient).size() == 0) {
hasFailedAlready = true;
throw new RuntimeException("Kafka server not available");
}
@@ -166,16 +166,16 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
@Override
protected void cleanupResource() throws Exception {
if (embedded) {
if (this.embedded) {
try {
kafkaServer.shutdown();
this.kafkaServer.shutdown();
}
catch (Exception e) {
// ignore errors on shutdown
log.error(e.getMessage(), e);
}
try {
Utils.rm(kafkaServer.config().logDirs());
Utils.rm(this.kafkaServer.config().logDirs());
}
catch (Exception e) {
// ignore errors on shutdown
@@ -183,15 +183,15 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
}
}
try {
zkClient.close();
this.zkClient.close();
}
catch (ZkInterruptedException e) {
// ignore errors on shutdown
log.error(e.getMessage(), e);
}
if (embedded) {
if (this.embedded) {
try {
zookeeper.shutdown();
this.zookeeper.shutdown();
}
catch (Exception e) {
// ignore errors on shutdown

55
spring-cloud-stream-binder-kafka/pom.xml
View File

@@ -14,13 +14,18 @@
</parent>
<properties>
<kafka.version>0.8.2.2</kafka.version>
<curator.version>2.6.0</curator.version>
<spring-integration-kafka.version>1.3.1.BUILD-SNAPSHOT</spring-integration-kafka.version>
<kafka.version>0.9.0.1</kafka.version>
<spring-kafka.version>1.0.3.BUILD-SNAPSHOT</spring-kafka.version>
<spring-integration-kafka.version>2.0.1.BUILD-SNAPSHOT</spring-integration-kafka.version>
<rxjava-math.version>1.0.0</rxjava-math.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
@@ -61,9 +66,20 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>${spring-kafka.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
<version>${spring-kafka.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<artifactId>kafka_2.11</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
@@ -78,29 +94,19 @@
<artifactId>rxjava-math</artifactId>
<version>${rxjava-math.version}</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<artifactId>kafka_2.11</artifactId>
<classifier>test</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<artifactId>kafka_2.11</artifactId>
<version>${kafka.version}</version>
<exclusions>
<exclusion>
@@ -115,7 +121,7 @@
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<artifactId>kafka_2.11</artifactId>
<classifier>test</classifier>
<version>${kafka.version}</version>
</dependency>
@@ -124,21 +130,6 @@
<artifactId>kafka-clients</artifactId>
<version>${kafka.version}</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>${curator.version}</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>${curator.version}</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<version>${curator.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
</project>

95
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
View File

@@ -1,95 +0,0 @@
/*
* 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.cloud.stream.binder.kafka;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import kafka.cluster.Broker;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils$;
import org.I0Itec.zkclient.ZkClient;
import scala.collection.JavaConversions;
import scala.collection.Seq;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
import org.springframework.integration.kafka.core.BrokerAddress;
import org.springframework.integration.kafka.core.Partition;
/**
* Health indicator for Kafka.
*
* @author Ilayaperumal Gopinathan
*/
public class KafkaBinderHealthIndicator implements HealthIndicator {
private final KafkaMessageChannelBinder binder;
private final KafkaBinderConfigurationProperties configurationProperties;
public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
KafkaBinderConfigurationProperties configurationProperties) {
this.binder = binder;
this.configurationProperties = configurationProperties;
}
@Override
public Health health() {
ZkClient zkClient = null;
try {
zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
configurationProperties.getZkSessionTimeout(),
configurationProperties.getZkConnectionTimeout(), ZKStringSerializer$.MODULE$);
Set<String> brokersInClusterSet = new HashSet<>();
Seq<Broker> allBrokersInCluster = ZkUtils$.MODULE$.getAllBrokersInCluster(zkClient);
Collection<Broker> brokersInCluster = JavaConversions.asJavaCollection(allBrokersInCluster);
for (Broker broker : brokersInCluster) {
brokersInClusterSet.add(broker.connectionString());
}
Set<String> downMessages = new HashSet<>();
for (Map.Entry<String, Collection<Partition>> entry : binder.getTopicsInUse().entrySet()) {
for (Partition partition : entry.getValue()) {
BrokerAddress address = binder.getConnectionFactory().getLeader(partition);
if (!brokersInClusterSet.contains(address.toString())) {
downMessages.add(address.toString());
}
}
}
if (downMessages.isEmpty()) {
return Health.up().build();
}
return Health.down().withDetail("Following brokers are down: ", downMessages.toString()).build();
}
catch (Exception e) {
return Health.down(e).build();
}
finally {
if (zkClient != null) {
try {
zkClient.close();
}
catch (Exception e) {
// ignore
}
}
}
}
}

30
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java
View File

@@ -21,13 +21,15 @@ package org.springframework.cloud.stream.binder.kafka;
*/
public class KafkaConsumerProperties {
private boolean autoRebalanceEnabled = true;
private boolean autoCommitOffset = true;
private Boolean autoCommitOnError;
private boolean resetOffsets;
private KafkaMessageChannelBinder.StartOffset startOffset;
private StartOffset startOffset;
private boolean enableDlq;
@@ -49,11 +51,11 @@ public class KafkaConsumerProperties {
this.resetOffsets = resetOffsets;
}
public KafkaMessageChannelBinder.StartOffset getStartOffset() {
public StartOffset getStartOffset() {
return startOffset;
}
public void setStartOffset(KafkaMessageChannelBinder.StartOffset startOffset) {
public void setStartOffset(StartOffset startOffset) {
this.startOffset = startOffset;
}
@@ -80,4 +82,26 @@ public class KafkaConsumerProperties {
public void setRecoveryInterval(int recoveryInterval) {
this.recoveryInterval = recoveryInterval;
}
public boolean isAutoRebalanceEnabled() {
return autoRebalanceEnabled;
}
public void setAutoRebalanceEnabled(boolean autoRebalanceEnabled) {
this.autoRebalanceEnabled = autoRebalanceEnabled;
}
public enum StartOffset {
earliest(-2L), latest(-1L);
private final long referencePoint;
StartOffset(long referencePoint) {
this.referencePoint = referencePoint;
}
public long getReferencePoint() {
return referencePoint;
}
}
}

13
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java
View File

@@ -26,12 +26,13 @@ import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
* @author Marius Bogoevici
*/
@ConfigurationProperties("spring.cloud.stream.kafka")
public class KafkaExtendedBindingProperties implements ExtendedBindingProperties<KafkaConsumerProperties, KafkaProducerProperties> {
public class KafkaExtendedBindingProperties
implements ExtendedBindingProperties<KafkaConsumerProperties, KafkaProducerProperties> {
private Map<String, KafkaBindingProperties> bindings = new HashMap<>();
public Map<String, KafkaBindingProperties> getBindings() {
return bindings;
return this.bindings;
}
public void setBindings(Map<String, KafkaBindingProperties> bindings) {
@@ -40,8 +41,8 @@ public class KafkaExtendedBindingProperties implements ExtendedBindingProperties
@Override
public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) {
if (bindings.containsKey(channelName) && bindings.get(channelName).getConsumer() != null) {
return bindings.get(channelName).getConsumer();
if (this.bindings.containsKey(channelName) && this.bindings.get(channelName).getConsumer() != null) {
return this.bindings.get(channelName).getConsumer();
}
else {
return new KafkaConsumerProperties();
@@ -50,8 +51,8 @@ public class KafkaExtendedBindingProperties implements ExtendedBindingProperties
@Override
public KafkaProducerProperties getExtendedProducerProperties(String channelName) {
if (bindings.containsKey(channelName) && bindings.get(channelName).getProducer() != null) {
return bindings.get(channelName).getProducer();
if (this.bindings.containsKey(channelName) && this.bindings.get(channelName).getProducer() != null) {
return this.bindings.get(channelName).getProducer();
}
else {
return new KafkaProducerProperties();

552
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -16,33 +16,32 @@
package org.springframework.cloud.stream.binder.kafka;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import kafka.admin.AdminUtils;
import kafka.api.OffsetRequest;
import kafka.api.TopicMetadata;
import kafka.common.ErrorMapping;
import kafka.serializer.DefaultDecoder;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.utils.Utils;
import scala.collection.Seq;
@@ -56,37 +55,29 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties;
import org.springframework.context.Lifecycle;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.kafka.core.ConnectionFactory;
import org.springframework.integration.kafka.core.DefaultConnectionFactory;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.core.ZookeeperConfiguration;
import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
import org.springframework.integration.kafka.listener.AcknowledgingMessageListener;
import org.springframework.integration.kafka.listener.Acknowledgment;
import org.springframework.integration.kafka.listener.ErrorHandler;
import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.integration.kafka.listener.KafkaNativeOffsetManager;
import org.springframework.integration.kafka.listener.MessageListener;
import org.springframework.integration.kafka.listener.OffsetManager;
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.ProducerListener;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.messaging.Message;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
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.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryOperations;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
@@ -104,31 +95,17 @@ import org.springframework.util.StringUtils;
*/
public class KafkaMessageChannelBinder extends
AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,
ExtendedProducerProperties<KafkaProducerProperties>, Collection<Partition>>
ExtendedProducerProperties<KafkaProducerProperties>, Collection<PartitionInfo>>
implements ExtendedPropertiesBinder<MessageChannel, KafkaConsumerProperties, KafkaProducerProperties>,
DisposableBean {
private static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer();
private static final ThreadFactory DAEMON_THREAD_FACTORY;
static {
CustomizableThreadFactory threadFactory = new CustomizableThreadFactory("kafka-binder-");
threadFactory.setDaemon(true);
DAEMON_THREAD_FACTORY = threadFactory;
}
private final KafkaBinderConfigurationProperties configurationProperties;
private RetryOperations metadataRetryOperations;
private final Map<String, Collection<Partition>> topicsInUse = new HashMap<>();
private final Map<String, Collection<PartitionInfo>> topicsInUse = new HashMap<>();
// -------- Default values for properties -------
private ConnectionFactory connectionFactory;
private ProducerListener producerListener;
private ProducerListener<byte[], byte[]> producerListener;
private volatile Producer<byte[], byte[]> dlqProducer;
@@ -155,14 +132,6 @@ public class KafkaMessageChannelBinder extends
return headersToMap;
}
ConnectionFactory getConnectionFactory() {
return this.connectionFactory;
}
public void setProducerListener(ProducerListener producerListener) {
this.producerListener = producerListener;
}
/**
* Retry configuration for operations such as validating topic creation
* @param metadataRetryOperations the retry configuration
@@ -177,13 +146,7 @@ public class KafkaMessageChannelBinder extends
@Override
public void onInit() throws Exception {
ZookeeperConfiguration configuration = new ZookeeperConfiguration(
new ZookeeperConnect(this.configurationProperties.getZkConnectionString()));
configuration.setBufferSize(this.configurationProperties.getSocketBufferSize());
configuration.setMaxWait(this.configurationProperties.getMaxWait());
DefaultConnectionFactory defaultConnectionFactory = new DefaultConnectionFactory(configuration);
defaultConnectionFactory.afterPropertiesSet();
this.connectionFactory = defaultConnectionFactory;
if (this.metadataRetryOperations == null) {
RetryTemplate retryTemplate = new RetryTemplate();
@@ -208,23 +171,12 @@ public class KafkaMessageChannelBinder extends
}
}
/**
* Allowed chars are ASCII alphanumerics, '.', '_' and '-'.
*/
static void validateTopicName(String topicName) {
try {
byte[] utf8 = topicName.getBytes("UTF-8");
for (byte b : utf8) {
if (!((b >= 'a') && (b <= 'z') || (b >= 'A') && (b <= 'Z') || (b >= '0') && (b <= '9') || (b == '.')
|| (b == '-') || (b == '_'))) {
throw new IllegalArgumentException(
"Topic name can only have ASCII alphanumerics, '.', '_' and '-'");
}
}
}
catch (UnsupportedEncodingException e) {
throw new AssertionError(e); // Can't happen
}
public void setProducerListener(ProducerListener<byte[], byte[]> producerListener) {
this.producerListener = producerListener;
}
Map<String, Collection<PartitionInfo>> getTopicsInUse() {
return this.topicsInUse;
}
@Override
@@ -237,30 +189,89 @@ public class KafkaMessageChannelBinder extends
return this.extendedBindingProperties.getExtendedProducerProperties(channelName);
}
Map<String, Collection<Partition>> getTopicsInUse() {
return this.topicsInUse;
@Override
protected MessageHandler createProducerMessageHandler(final String name,
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {
KafkaTopicUtils.validateTopicName(name);
Collection<PartitionInfo> partitions = ensureTopicCreated(name, producerProperties.getPartitionCount());
if (producerProperties.getPartitionCount() < partitions.size()) {
if (this.logger.isInfoEnabled()) {
this.logger.info("The `partitionCount` of the producer for topic " + name + " is "
+ producerProperties.getPartitionCount() + ", smaller than the actual partition count of "
+ partitions.size() + " of the topic. The larger number will be used instead.");
}
}
this.topicsInUse.put(name, partitions);
ProducerFactory<byte[], byte[]> producerFB = getProducerFactory(producerProperties);
KafkaTemplate<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFB);
if (this.producerListener != null) {
kafkaTemplate.setProducerListener(this.producerListener);
}
return new ProducerConfigurationMessageHandler(kafkaTemplate, name, producerProperties);
}
@Override
protected Collection<Partition> createConsumerDestinationIfNecessary(String name, String group,
protected void createProducerDestinationIfNecessary(String name,
ExtendedProducerProperties<KafkaProducerProperties> properties) {
if (this.logger.isInfoEnabled()) {
this.logger.info("Using kafka topic for outbound: " + name);
}
KafkaTopicUtils.validateTopicName(name);
Collection<PartitionInfo> partitions = ensureTopicCreated(name, properties.getPartitionCount());
if (properties.getPartitionCount() < partitions.size()) {
if (this.logger.isInfoEnabled()) {
this.logger.info("The `partitionCount` of the producer for topic " + name + " is "
+ properties.getPartitionCount() + ", smaller than the actual partition count of "
+ partitions.size() + " of the topic. The larger number will be used instead.");
}
}
this.topicsInUse.put(name, partitions);
}
private ProducerFactory<byte[], byte[]> getProducerFactory(
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
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, ByteArraySerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks()));
props.put(ProducerConfig.LINGER_MS_CONFIG,
String.valueOf(producerProperties.getExtension().getBatchTimeout()));
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
producerProperties.getExtension().getCompressionType().toString());
return new DefaultKafkaProducerFactory<>(props);
}
@Override
protected Collection<PartitionInfo> createConsumerDestinationIfNecessary(String name, String group,
ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
validateTopicName(name);
KafkaTopicUtils.validateTopicName(name);
if (properties.getInstanceCount() == 0) {
throw new IllegalArgumentException("Instance count cannot be zero");
}
Collection<Partition> allPartitions = ensureTopicCreated(name,
Collection<PartitionInfo> allPartitions = ensureTopicCreated(name,
properties.getInstanceCount() * properties.getConcurrency());
Collection<Partition> listenedPartitions;
Collection<PartitionInfo> listenedPartitions;
if (properties.getInstanceCount() == 1) {
listenedPartitions = allPartitions;
}
else {
listenedPartitions = new ArrayList<>();
for (Partition partition : allPartitions) {
for (PartitionInfo partition : allPartitions) {
// divide partitions across modules
if ((partition.getId() % properties.getInstanceCount()) == properties.getInstanceIndex()) {
if ((partition.partition() % properties.getInstanceCount()) == properties.getInstanceIndex()) {
listenedPartitions.add(partition);
}
}
@@ -269,135 +280,73 @@ public class KafkaMessageChannelBinder extends
return listenedPartitions;
}
@Override
@SuppressWarnings("unchecked")
protected MessageProducer createConsumerEndpoint(String name, String group, Collection<Partition> destination,
protected MessageProducer createConsumerEndpoint(String name, String group, Collection<PartitionInfo> destination,
ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
Assert.isTrue(!CollectionUtils.isEmpty(destination), "A list of partitions must be provided");
int concurrency = Math.min(properties.getConcurrency(), destination.size());
final ExecutorService dispatcherTaskExecutor =
Executors.newFixedThreadPool(concurrency, DAEMON_THREAD_FACTORY);
final KafkaMessageListenerContainer messageListenerContainer = new KafkaMessageListenerContainer(
this.connectionFactory, destination.toArray(new Partition[destination.size()])) {
@Override
public void stop(Runnable callback) {
super.stop(callback);
if (getOffsetManager() instanceof DisposableBean) {
try {
((DisposableBean) getOffsetManager()).destroy();
}
catch (Exception e) {
KafkaMessageChannelBinder.this.logger.error("Error while closing the offset manager", e);
}
}
dispatcherTaskExecutor.shutdown();
}
};
if (this.logger.isDebugEnabled()) {
this.logger.debug(
"Listened partitions: " + StringUtils.collectionToCommaDelimitedString(destination));
}
boolean anonymous = !StringUtils.hasText(group);
Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(),
"DLQ support is not available for anonymous subscriptions");
String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group;
long referencePoint = properties.getExtension().getStartOffset() != null
? properties.getExtension().getStartOffset().getReferencePoint()
: (anonymous ? OffsetRequest.LatestTime() : OffsetRequest.EarliestTime());
OffsetManager offsetManager = createOffsetManager(consumerGroup, referencePoint);
if (properties.getExtension().isResetOffsets()) {
offsetManager.resetOffsets(destination);
}
messageListenerContainer.setOffsetManager(offsetManager);
messageListenerContainer.setQueueSize(this.configurationProperties.getQueueSize());
messageListenerContainer.setMaxFetch(this.configurationProperties.getFetchSize());
boolean autoCommitOnError = properties.getExtension().getAutoCommitOnError() != null
? properties.getExtension().getAutoCommitOnError()
: properties.getExtension().isAutoCommitOffset() && properties.getExtension().isEnableDlq();
messageListenerContainer.setAutoCommitOnError(autoCommitOnError);
messageListenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
Map<String, Object> props = getConsumerConfig(anonymous, consumerGroup);
Deserializer<byte[]> valueDecoder = new ByteArrayDeserializer();
Deserializer<byte[]> keyDecoder = new ByteArrayDeserializer();
ConsumerFactory<byte[], byte[]> consumerFactory = new DefaultKafkaConsumerFactory<>(props, keyDecoder,
valueDecoder);
Collection<PartitionInfo> listenedPartitions = (Collection<PartitionInfo>) destination;
Assert.isTrue(!CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided");
final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets(
listenedPartitions);
final ContainerProperties containerProperties =
anonymous || properties.getExtension().isAutoRebalanceEnabled() ? new ContainerProperties(name)
: new ContainerProperties(topicPartitionInitialOffsets);
int concurrency = Math.min(properties.getConcurrency(), listenedPartitions.size());
final ConcurrentMessageListenerContainer<byte[], byte[]> messageListenerContainer =
new ConcurrentMessageListenerContainer(
consumerFactory, containerProperties) {
@Override
public void stop(Runnable callback) {
super.stop(callback);
}
};
messageListenerContainer.setConcurrency(concurrency);
messageListenerContainer.setDispatcherTaskExecutor(dispatcherTaskExecutor);
final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter(
messageListenerContainer);
kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());
kafkaMessageDrivenChannelAdapter.setKeyDecoder(new DefaultDecoder(null));
kafkaMessageDrivenChannelAdapter.setPayloadDecoder(new DefaultDecoder(null));
kafkaMessageDrivenChannelAdapter.setAutoCommitOffset(properties.getExtension().isAutoCommitOffset());
kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
if (properties.getMaxAttempts() > 1) {
// we need to wrap the adapter listener into a retrying listener so that the retry
// logic is applied before the ErrorHandler is executed
final RetryTemplate retryTemplate = buildRetryTemplate(properties);
if (properties.getExtension().isAutoCommitOffset()) {
final MessageListener originalMessageListener = (MessageListener) messageListenerContainer
.getMessageListener();
messageListenerContainer.setMessageListener(new MessageListener() {
messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(properties));
@Override
public void onMessage(final KafkaMessage message) {
try {
retryTemplate.execute(new RetryCallback<Object, Throwable>() {
@Override
public Object doWithRetry(RetryContext context) {
originalMessageListener.onMessage(message);
return null;
}
});
}
catch (Throwable throwable) {
if (throwable instanceof RuntimeException) {
throw (RuntimeException) throwable;
}
else {
throw new RuntimeException(throwable);
}
}
}
});
}
else {
messageListenerContainer.setMessageListener(new AcknowledgingMessageListener() {
final AcknowledgingMessageListener originalMessageListener =
(AcknowledgingMessageListener) messageListenerContainer
.getMessageListener();
@Override
public void onMessage(final KafkaMessage message, final Acknowledgment acknowledgment) {
retryTemplate.execute(new RetryCallback<Object, RuntimeException>() {
@Override
public Object doWithRetry(RetryContext context) {
originalMessageListener.onMessage(message, acknowledgment);
return null;
}
});
}
});
}
if (this.logger.isDebugEnabled()) {
this.logger.debug(
"Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions));
}
if (this.logger.isDebugEnabled()) {
this.logger.debug(
"Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions));
}
final KafkaMessageDrivenChannelAdapter<byte[], byte[]> kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter<>(
messageListenerContainer);
kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());
final RetryTemplate retryTemplate = buildRetryTemplate(properties);
kafkaMessageDrivenChannelAdapter.setRetryTemplate(retryTemplate);
if (properties.getExtension().isEnableDlq()) {
final String dlqTopic = "error." + name + "." + consumerGroup;
final String dlqTopic = "error." + name + "." + group;
initDlqProducer();
messageListenerContainer.setErrorHandler(new ErrorHandler() {
messageListenerContainer.getContainerProperties().setErrorHandler(new ErrorHandler() {
@Override
public void handle(Exception thrownException, final KafkaMessage message) {
final byte[] key = message.getMessage().key() != null ? Utils.toArray(message.getMessage().key())
public void handle(Exception thrownException, final ConsumerRecord message) {
final byte[] key = message.key() != null ? Utils.toArray(ByteBuffer.wrap((byte[]) message.key()))
: null;
final byte[] payload = message.getMessage().payload() != null
? Utils.toArray(message.getMessage().payload()) : null;
final byte[] payload = message.value() != null
? Utils.toArray(ByteBuffer.wrap((byte[]) message.value())) : null;
KafkaMessageChannelBinder.this.dlqProducer.send(new ProducerRecord<>(dlqTopic, key, payload),
new Callback() {
@@ -408,7 +357,7 @@ public class KafkaMessageChannelBinder extends
+ toDisplayString(ObjectUtils.nullSafeToString(key), 50) + "'");
messageLog.append(" and payload='"
+ toDisplayString(ObjectUtils.nullSafeToString(payload), 50) + "'");
messageLog.append(" received from " + message.getMetadata().getPartition());
messageLog.append(" received from " + message.partition());
if (exception != null) {
KafkaMessageChannelBinder.this.logger.error(
"Error sending to DLQ" + messageLog.toString(), exception);
@@ -427,74 +376,60 @@ public class KafkaMessageChannelBinder extends
return kafkaMessageDrivenChannelAdapter;
}
@Override
protected MessageHandler createProducerMessageHandler(final String destination,
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {
ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<>(destination, byte[].class,
byte[].class,
BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER);
producerMetadata.setSync(producerProperties.getExtension().isSync());
producerMetadata.setCompressionType(producerProperties.getExtension().getCompressionType());
producerMetadata.setBatchBytes(producerProperties.getExtension().getBufferSize());
Properties additional = new Properties();
additional.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks()));
additional.put(ProducerConfig.LINGER_MS_CONFIG,
String.valueOf(producerProperties.getExtension().getBatchTimeout()));
ProducerFactoryBean<byte[], byte[]> producerFB = new ProducerFactoryBean<>(producerMetadata,
this.configurationProperties.getKafkaConnectionString(), additional);
final ProducerConfiguration<byte[], byte[]> producerConfiguration = new ProducerConfiguration<>(
producerMetadata, producerFB.getObject());
producerConfiguration.setProducerListener(this.producerListener);
KafkaProducerContext kafkaProducerContext = new KafkaProducerContext();
kafkaProducerContext.setProducerConfigurations(
Collections.<String, ProducerConfiguration<?, ?>>singletonMap(destination, producerConfiguration));
return new ProducerConfigurationMessageHandler(producerConfiguration, destination);
private Map<String, Object> getConsumerConfig(boolean anonymous, String consumerGroup) {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
anonymous ? "latest" : "earliest");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
return props;
}
@Override
protected void createProducerDestinationIfNecessary(String name,
ExtendedProducerProperties<KafkaProducerProperties> properties) {
if (this.logger.isInfoEnabled()) {
this.logger.info("Using kafka topic for outbound: " + name);
private boolean isAutoCommitOnError(ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
return properties.getExtension().getAutoCommitOnError() != null
? properties.getExtension().getAutoCommitOnError()
: properties.getExtension().isAutoCommitOffset() && properties.getExtension().isEnableDlq();
}
private TopicPartitionInitialOffset[] getTopicPartitionInitialOffsets(
Collection<PartitionInfo> listenedPartitions) {
final TopicPartitionInitialOffset[] topicPartitionInitialOffsets =
new TopicPartitionInitialOffset[listenedPartitions.size()];
int i = 0;
for (PartitionInfo partition : listenedPartitions) {
topicPartitionInitialOffsets[i++] = new TopicPartitionInitialOffset(partition.topic(),
partition.partition());
}
validateTopicName(name);
Collection<Partition> partitions = ensureTopicCreated(name, properties.getPartitionCount());
// If the topic already exists, and it has a larger number of partitions than the one set in `partitionCount`,
// we will use the existing partition count of the topic instead of the user setting.
if (properties.getPartitionCount() < partitions.size()) {
if (this.logger.isInfoEnabled()) {
this.logger.info("The `partitionCount` setting of the producer for topic " + name + " is "
+ properties.getPartitionCount() + ", smaller than the actual partition count of "
+ partitions.size() + " of the topic. The larger number will be used instead.");
}
}
this.topicsInUse.put(name, partitions);
return topicPartitionInitialOffsets;
}
/**
* Creates a Kafka topic if needed, or try to increase its partition count to the
* desired number. If a topic with a larger number of partitions already exists,
* the partition count remains unchanged.
* desired number.
*/
private Collection<Partition> ensureTopicCreated(final String topicName, final int partitionCount) {
private Collection<PartitionInfo> ensureTopicCreated(final String topicName, final int partitionCount) {
final ZkClient zkClient = new ZkClient(this.configurationProperties.getZkConnectionString(),
this.configurationProperties.getZkSessionTimeout(),
this.configurationProperties.getZkConnectionTimeout(),
ZKStringSerializer$.MODULE$);
final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
try {
final Properties topicConfig = new Properties();
TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkClient);
TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkUtils);
if (topicMetadata.errorCode() == ErrorMapping.NoError()) {
// only consider minPartitionCount for resizing if autoAddPartitions is
// true
int effectivePartitionCount = this.configurationProperties.isAutoAddPartitions()
? Math.max(this.configurationProperties.getMinPartitionCount(),
partitionCount) : partitionCount;
? Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount)
: partitionCount;
if (topicMetadata.partitionsMetadata().size() < effectivePartitionCount) {
if (this.configurationProperties.isAutoAddPartitions()) {
AdminUtils.addPartitions(zkClient, topicName, effectivePartitionCount, null, false,
new Properties());
AdminUtils.addPartitions(zkUtils, topicName, effectivePartitionCount, null, false);
}
else {
int topicSize = topicMetadata.partitionsMetadata().size();
@@ -507,7 +442,7 @@ public class KafkaMessageChannelBinder extends
}
else if (topicMetadata.errorCode() == ErrorMapping.UnknownTopicOrPartitionCode()) {
if (this.configurationProperties.isAutoCreateTopics()) {
Seq<Object> brokerList = ZkUtils.getSortedBrokerList(zkClient);
Seq<Object> brokerList = zkUtils.getSortedBrokerList();
// always consider minPartitionCount for topic creation
int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(),
partitionCount);
@@ -518,7 +453,7 @@ public class KafkaMessageChannelBinder extends
@Override
public Object doWithRetry(RetryContext context) throws RuntimeException {
AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName,
AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topicName,
replicaAssignment, topicConfig, true);
return null;
}
@@ -533,26 +468,25 @@ public class KafkaMessageChannelBinder extends
ErrorMapping.exceptionFor(topicMetadata.errorCode()));
}
try {
Collection<Partition> partitions = this.metadataRetryOperations
.execute(new RetryCallback<Collection<Partition>, Exception>() {
return this.metadataRetryOperations
.execute(new RetryCallback<Collection<PartitionInfo>, Exception>() {
@Override
public Collection<Partition> doWithRetry(RetryContext context) throws Exception {
KafkaMessageChannelBinder.this.connectionFactory.refreshMetadata(
Collections.singleton(topicName));
Collection<Partition> partitions =
KafkaMessageChannelBinder.this.connectionFactory.getPartitions(topicName);
public Collection<PartitionInfo> doWithRetry(RetryContext context) throws Exception {
Collection<PartitionInfo> partitions =
getProducerFactory(
new ExtendedProducerProperties<>(new KafkaProducerProperties()))
.createProducer().partitionsFor(topicName);
// do a sanity check on the partition set
if (partitions.size() < partitionCount) {
throw new IllegalStateException("The number of expected partitions was: "
+ partitionCount + ", but " + partitions.size()
+ (partitions.size() > 1 ? " have " : " has ") + "been found instead");
}
KafkaMessageChannelBinder.this.connectionFactory.getLeaders(partitions);
return partitions;
}
});
return partitions;
}
catch (Exception e) {
this.logger.error("Cannot initialize Binder", e);
@@ -572,19 +506,18 @@ public class KafkaMessageChannelBinder extends
if (this.dlqProducer == null) {
// we can use the producer defaults as we do not need to tune
// performance
ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<>("dlqKafkaProducer",
byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER);
producerMetadata.setSync(false);
producerMetadata.setCompressionType(ProducerMetadata.CompressionType.none);
producerMetadata.setBatchBytes(16384);
Properties additionalProps = new Properties();
additionalProps.put(ProducerConfig.ACKS_CONFIG,
String.valueOf(this.configurationProperties.getRequiredAcks()));
additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(0));
ProducerFactoryBean<byte[], byte[]> producerFactoryBean = new ProducerFactoryBean<>(
producerMetadata, this.configurationProperties.getKafkaConnectionString(),
additionalProps);
this.dlqProducer = producerFactoryBean.getObject();
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
this.configurationProperties.getKafkaConnectionString());
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, ByteArraySerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
DefaultKafkaProducerFactory<byte[], byte[]> defaultKafkaProducerFactory =
new DefaultKafkaProducerFactory<>(props);
this.dlqProducer = defaultKafkaProducerFactory.createProducer();
}
}
}
@@ -594,29 +527,6 @@ public class KafkaMessageChannelBinder extends
}
}
private OffsetManager createOffsetManager(String group, long referencePoint) {
try {
KafkaNativeOffsetManager kafkaOffsetManager = new KafkaNativeOffsetManager(this.connectionFactory,
new ZookeeperConnect(this.configurationProperties.getZkConnectionString()),
Collections.<Partition, Long>emptyMap());
kafkaOffsetManager.setConsumerId(group);
kafkaOffsetManager.setReferenceTimestamp(referencePoint);
kafkaOffsetManager.afterPropertiesSet();
WindowingOffsetManager windowingOffsetManager = new WindowingOffsetManager(kafkaOffsetManager);
windowingOffsetManager.setTimespan(this.configurationProperties.getOffsetUpdateTimeWindow());
windowingOffsetManager.setCount(this.configurationProperties.getOffsetUpdateCount());
windowingOffsetManager.setShutdownTimeout(this.configurationProperties.getOffsetUpdateShutdownTimeout());
windowingOffsetManager.afterPropertiesSet();
return windowingOffsetManager;
}
catch (Exception e) {
throw new RuntimeException(e);
}
}
private String toDisplayString(String original, int maxCharacters) {
if (original.length() <= maxCharacters) {
return original;
@@ -624,45 +534,38 @@ public class KafkaMessageChannelBinder extends
return original.substring(0, maxCharacters) + "...";
}
public enum StartOffset {
earliest(OffsetRequest.EarliestTime()), latest(OffsetRequest.LatestTime());
private final class ProducerConfigurationMessageHandler extends KafkaProducerMessageHandler<byte[], byte[]>
implements Lifecycle {
private final long referencePoint;
StartOffset(long referencePoint) {
this.referencePoint = referencePoint;
}
public long getReferencePoint() {
return this.referencePoint;
}
}
private final static class ProducerConfigurationMessageHandler implements MessageHandler, Lifecycle {
private ProducerConfiguration<byte[], byte[]> delegate;
private String targetTopic;
private boolean running;
private ProducerConfigurationMessageHandler(
ProducerConfiguration<byte[], byte[]> delegate, String targetTopic) {
Assert.notNull(delegate, "Delegate cannot be null");
Assert.hasText(targetTopic, "Target topic cannot be null");
this.delegate = delegate;
this.targetTopic = targetTopic;
private boolean running = true;
private ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic,
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
super(kafkaTemplate);
setTopicExpression(new LiteralExpression(topic));
setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
if (producerProperties.isPartitioned()) {
SpelExpressionParser parser = new SpelExpressionParser();
setPartitionIdExpression(parser.parseExpression("headers.partition"));
}
if (producerProperties.getExtension().isSync()) {
setSync(true);
}
}
@Override
public void start() {
this.running = true;
try {
super.onInit();
}
catch (Exception e) {
this.logger.error("Initialization errors: ", e);
throw new RuntimeException(e);
}
}
@Override
public void stop() {
this.delegate.stop();
this.running = false;
}
@@ -670,12 +573,5 @@ public class KafkaMessageChannelBinder extends
public boolean isRunning() {
return this.running;
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
this.delegate.send(this.targetTopic,
message.getHeaders().get(BinderHeaders.PARTITION_HEADER, Integer.class), null,
(byte[]) message.getPayload());
}
}
}

22
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaProducerProperties.java
View File

@@ -18,8 +18,6 @@ package org.springframework.cloud.stream.binder.kafka;
import javax.validation.constraints.NotNull;
import org.springframework.integration.kafka.support.ProducerMetadata;
/**
* @author Marius Bogoevici
*/
@@ -27,14 +25,14 @@ public class KafkaProducerProperties {
private int bufferSize = 16384;
private ProducerMetadata.CompressionType compressionType = ProducerMetadata.CompressionType.none;
private CompressionType compressionType = CompressionType.none;
private boolean sync;
private int batchTimeout;
public int getBufferSize() {
return bufferSize;
return this.bufferSize;
}
public void setBufferSize(int bufferSize) {
@@ -42,16 +40,16 @@ public class KafkaProducerProperties {
}
@NotNull
public ProducerMetadata.CompressionType getCompressionType() {
return compressionType;
public CompressionType getCompressionType() {
return this.compressionType;
}
public void setCompressionType(ProducerMetadata.CompressionType compressionType) {
public void setCompressionType(CompressionType compressionType) {
this.compressionType = compressionType;
}
public boolean isSync() {
return sync;
return this.sync;
}
public void setSync(boolean sync) {
@@ -59,10 +57,16 @@ public class KafkaProducerProperties {
}
public int getBatchTimeout() {
return batchTimeout;
return this.batchTimeout;
}
public void setBatchTimeout(int batchTimeout) {
this.batchTimeout = batchTimeout;
}
public enum CompressionType {
none,
gzip,
snappy
}
}

48
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaTopicUtils.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* 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.cloud.stream.binder.kafka;
import java.io.UnsupportedEncodingException;
/**
* @author Soby Chacko
*/
public final class KafkaTopicUtils {
private KafkaTopicUtils() {
}
/**
* Allowed chars are ASCII alphanumerics, '.', '_' and '-'.
*/
public static void validateTopicName(String topicName) {
try {
byte[] utf8 = topicName.getBytes("UTF-8");
for (byte b : utf8) {
if (!((b >= 'a') && (b <= 'z') || (b >= 'A') && (b <= 'Z') || (b >= '0') && (b <= '9') || (b == '.')
|| (b == '-') || (b == '_'))) {
throw new IllegalArgumentException(
"Topic name can only have ASCII alphanumerics, '.', '_' and '-'");
}
}
}
catch (UnsupportedEncodingException e) {
throw new AssertionError(e); // Can't happen
}
}
}

262
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/WindowingOffsetManager.java
View File

@@ -1,262 +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.cloud.stream.binder.kafka;
import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import rx.Observable;
import rx.Subscription;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.observables.GroupedObservable;
import rx.observables.MathObservable;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.listener.OffsetManager;
import org.springframework.util.Assert;
/**
* An {@link OffsetManager} that aggregates writes over a time or count window, using an underlying delegate to
* do the actual operations. Its purpose is to reduce the performance impact of writing operations
* wherever this is desirable.
*
* Either a time window or a number of writes can be specified, but not both.
*
* @author Marius Bogoevici
*/
public class WindowingOffsetManager implements OffsetManager, InitializingBean, DisposableBean {
private final CreatePartitionAndOffsetFunction createPartitionAndOffsetFunction = new CreatePartitionAndOffsetFunction();
private final GetOffsetFunction getOffsetFunction = new GetOffsetFunction();
private final ComputeMaximumOffsetByPartitionFunction findHighestOffsetInPartitionGroup = new ComputeMaximumOffsetByPartitionFunction();
private final GetPartitionFunction getPartition = new GetPartitionFunction();
private final FindHighestOffsetsByPartitionFunction findHighestOffsetsByPartition = new FindHighestOffsetsByPartitionFunction();
private final DelegateUpdateOffsetAction delegateUpdateOffsetAction = new DelegateUpdateOffsetAction();
private final NotifyObservableClosedAction notifyObservableClosed = new NotifyObservableClosedAction();
private final OffsetManager delegate;
private long timespan = 10 * 1000;
private int count;
private Subject<PartitionAndOffset, PartitionAndOffset> offsets;
private Subscription subscription;
private int shutdownTimeout = 2000;
private CountDownLatch shutdownLatch;
public WindowingOffsetManager(OffsetManager offsetManager) {
this.delegate = offsetManager;
}
/**
* The timespan for aggregating write operations, before invoking the underlying {@link OffsetManager}.
*
* @param timespan duration in milliseconds
*/
public void setTimespan(long timespan) {
Assert.isTrue(timespan >= 0, "Timespan must be a positive value");
this.timespan = timespan;
}
/**
* How many writes should be aggregated, before invoking the underlying {@link OffsetManager}. Setting this value
* to 1 effectively disables windowing.
*
* @param count number of writes
*/
public void setCount(int count) {
Assert.isTrue(count >= 0, "Count must be a positive value");
this.count = count;
}
/**
* The timeout that {@link #close()} and {@link #destroy()} operations will wait for receving a confirmation that the
* underlying writes have been processed.
*
* @param shutdownTimeout duration in milliseconds
*/
public void setShutdownTimeout(int shutdownTimeout) {
this.shutdownTimeout = shutdownTimeout;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.isTrue(timespan > 0 ^ count > 0, "Only one of the timespan or count must be set");
// create the stream if windowing is set, and count is higher than 1
if (timespan > 0 || count > 1) {
offsets = new SerializedSubject<>(PublishSubject.<PartitionAndOffset>create());
// window by either count or time
Observable<Observable<PartitionAndOffset>> window =
timespan > 0 ? offsets.window(timespan, TimeUnit.MILLISECONDS) : offsets.window(count);
Observable<PartitionAndOffset> maximumOffsetsByWindow = window
.flatMap(findHighestOffsetsByPartition)
.doOnCompleted(notifyObservableClosed);
subscription = maximumOffsetsByWindow.subscribe(delegateUpdateOffsetAction);
}
else {
offsets = null;
}
}
@Override
public void destroy() throws Exception {
this.flush();
this.close();
if (delegate instanceof DisposableBean) {
((DisposableBean) delegate).destroy();
}
}
@Override
public void updateOffset(Partition partition, long offset) {
if (offsets != null) {
offsets.onNext(new PartitionAndOffset(partition, offset));
}
else {
delegate.updateOffset(partition, offset);
}
}
@Override
public long getOffset(Partition partition) {
return delegate.getOffset(partition);
}
@Override
public void deleteOffset(Partition partition) {
delegate.deleteOffset(partition);
}
@Override
public void resetOffsets(Collection<Partition> partition) {
delegate.resetOffsets(partition);
}
@Override
public void close() throws IOException {
if (offsets != null) {
shutdownLatch = new CountDownLatch(1);
offsets.onCompleted();
try {
shutdownLatch.await(shutdownTimeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
// ignore
}
subscription.unsubscribe();
}
delegate.close();
}
@Override
public void flush() throws IOException {
delegate.flush();
}
private final class PartitionAndOffset {
private final Partition partition;
private final Long offset;
private PartitionAndOffset(Partition partition, Long offset) {
this.partition = partition;
this.offset = offset;
}
public Partition getPartition() {
return partition;
}
public Long getOffset() {
return offset;
}
}
private class DelegateUpdateOffsetAction implements Action1<PartitionAndOffset> {
@Override
public void call(PartitionAndOffset partitionAndOffset) {
delegate.updateOffset(partitionAndOffset.getPartition(), partitionAndOffset.getOffset());
}
}
private class NotifyObservableClosedAction implements Action0 {
@Override
public void call() {
if (shutdownLatch != null) {
shutdownLatch.countDown();
}
}
}
private class CreatePartitionAndOffsetFunction implements Func2<Partition, Long, PartitionAndOffset> {
@Override
public PartitionAndOffset call(Partition partition, Long offset) {
return new PartitionAndOffset(partition, offset);
}
}
private class GetOffsetFunction implements Func1<PartitionAndOffset, Long> {
@Override
public Long call(PartitionAndOffset partitionAndOffset) {
return partitionAndOffset.getOffset();
}
}
private class ComputeMaximumOffsetByPartitionFunction implements Func1<GroupedObservable<Partition, PartitionAndOffset>, Observable<PartitionAndOffset>> {
@Override
public Observable<PartitionAndOffset> call(GroupedObservable<Partition, PartitionAndOffset> group) {
return Observable.zip(Observable.just(group.getKey()),
MathObservable.max(group.map(getOffsetFunction)),
createPartitionAndOffsetFunction);
}
}
private class GetPartitionFunction implements Func1<PartitionAndOffset, Partition> {
@Override
public Partition call(PartitionAndOffset partitionAndOffset) {
return partitionAndOffset.getPartition();
}
}
private class FindHighestOffsetsByPartitionFunction implements Func1<Observable<PartitionAndOffset>, Observable<PartitionAndOffset>> {
@Override
public Observable<PartitionAndOffset> call(Observable<PartitionAndOffset> windowBuffer) {
return windowBuffer.groupBy(getPartition).flatMap(findHighestOffsetInPartitionGroup);
}
}
}

26
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
View File

@@ -21,7 +21,6 @@ import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfigurati
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator;
import org.springframework.cloud.stream.binder.kafka.KafkaExtendedBindingProperties;
import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
@@ -29,8 +28,8 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.kafka.support.LoggingProducerListener;
import org.springframework.integration.kafka.support.ProducerListener;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
/**
* @author David Turanski
@@ -41,8 +40,8 @@ import org.springframework.integration.kafka.support.ProducerListener;
*/
@Configuration
@ConditionalOnMissingBean(Binder.class)
@Import({ KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class })
@EnableConfigurationProperties({ KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class })
@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class})
public class KafkaBinderConfiguration {
@Autowired
@@ -59,10 +58,11 @@ public class KafkaBinderConfiguration {
@Bean
KafkaMessageChannelBinder kafkaMessageChannelBinder() {
KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(configurationProperties);
kafkaMessageChannelBinder.setCodec(codec);
kafkaMessageChannelBinder.setProducerListener(producerListener);
kafkaMessageChannelBinder.setExtendedBindingProperties(kafkaExtendedBindingProperties);
KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(
this.configurationProperties);
kafkaMessageChannelBinder.setCodec(this.codec);
//kafkaMessageChannelBinder.setProducerListener(producerListener);
kafkaMessageChannelBinder.setExtendedBindingProperties(this.kafkaExtendedBindingProperties);
return kafkaMessageChannelBinder;
}
@@ -72,8 +72,8 @@ public class KafkaBinderConfiguration {
return new LoggingProducerListener();
}
@Bean
KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) {
return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, configurationProperties);
}
// @Bean
// KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) {
// return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, configurationProperties);
// }
}

40
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java
View File

@@ -23,15 +23,16 @@ import org.springframework.util.StringUtils;
* @author David Turanski
* @author Ilayaperumal Gopinathan
* @author Marius Bogoevici
* @author Soby Chacko
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder")
public class KafkaBinderConfigurationProperties {
private String[] zkNodes = new String[] { "localhost" };
private String[] zkNodes = new String[] {"localhost"};
private String defaultZkPort = "2181";
private String[] brokers = new String[] { "localhost" };
private String[] brokers = new String[] {"localhost"};
private String defaultBrokerPort = "9092";
@@ -71,6 +72,12 @@ public class KafkaBinderConfigurationProperties {
private int queueSize = 8192;
private String consumerGroup;
public String getConsumerGroup() {
return this.consumerGroup;
}
public String getZkConnectionString() {
return toConnectionString(this.zkNodes, this.defaultZkPort);
}
@@ -80,7 +87,7 @@ public class KafkaBinderConfigurationProperties {
}
public String[] getHeaders() {
return headers;
return this.headers;
}
public int getOffsetUpdateTimeWindow() {
@@ -96,7 +103,7 @@ public class KafkaBinderConfigurationProperties {
}
public String[] getZkNodes() {
return zkNodes;
return this.zkNodes;
}
public void setZkNodes(String... zkNodes) {
@@ -108,7 +115,7 @@ public class KafkaBinderConfigurationProperties {
}
public String[] getBrokers() {
return brokers;
return this.brokers;
}
public void setBrokers(String... brokers) {
@@ -170,7 +177,7 @@ public class KafkaBinderConfigurationProperties {
}
public int getMaxWait() {
return maxWait;
return this.maxWait;
}
public void setMaxWait(int maxWait) {
@@ -178,7 +185,7 @@ public class KafkaBinderConfigurationProperties {
}
public int getRequiredAcks() {
return requiredAcks;
return this.requiredAcks;
}
public void setRequiredAcks(int requiredAcks) {
@@ -186,7 +193,7 @@ public class KafkaBinderConfigurationProperties {
}
public int getReplicationFactor() {
return replicationFactor;
return this.replicationFactor;
}
public void setReplicationFactor(int replicationFactor) {
@@ -194,7 +201,7 @@ public class KafkaBinderConfigurationProperties {
}
public int getFetchSize() {
return fetchSize;
return this.fetchSize;
}
public void setFetchSize(int fetchSize) {
@@ -202,7 +209,7 @@ public class KafkaBinderConfigurationProperties {
}
public int getMinPartitionCount() {
return minPartitionCount;
return this.minPartitionCount;
}
public void setMinPartitionCount(int minPartitionCount) {
@@ -210,7 +217,7 @@ public class KafkaBinderConfigurationProperties {
}
public int getQueueSize() {
return queueSize;
return this.queueSize;
}
public void setQueueSize(int queueSize) {
@@ -218,7 +225,7 @@ public class KafkaBinderConfigurationProperties {
}
public boolean isAutoCreateTopics() {
return autoCreateTopics;
return this.autoCreateTopics;
}
public void setAutoCreateTopics(boolean autoCreateTopics) {
@@ -226,7 +233,7 @@ public class KafkaBinderConfigurationProperties {
}
public boolean isAutoAddPartitions() {
return autoAddPartitions;
return this.autoAddPartitions;
}
public void setAutoAddPartitions(boolean autoAddPartitions) {
@@ -234,10 +241,15 @@ public class KafkaBinderConfigurationProperties {
}
public int getSocketBufferSize() {
return socketBufferSize;
return this.socketBufferSize;
}
public void setSocketBufferSize(int socketBufferSize) {
this.socketBufferSize = socketBufferSize;
}
public void setConsumerGroup(String consumerGroup) {
this.consumerGroup = consumerGroup;
}
}

766
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

File diff suppressed because it is too large Load Diff

4
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java
View File

@@ -27,9 +27,9 @@ import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.KryoRegistrar;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.kafka.support.LoggingProducerListener;
import org.springframework.integration.kafka.support.ProducerListener;
import org.springframework.integration.tuple.TupleKryoRegistrar;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Registration;

70
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java
View File

@@ -18,14 +18,12 @@ package org.springframework.cloud.stream.binder.kafka;
import java.util.Arrays;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.HeaderMode;
import org.springframework.context.Lifecycle;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -48,14 +46,15 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
@Override
public void testPartitionedModuleJava() throws Exception {
KafkaTestBinder binder = getBinder();
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
producerProperties.setHeaderMode(HeaderMode.raw);
producerProperties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class);
producerProperties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class);
producerProperties.setPartitionCount(6);
ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
properties.setHeaderMode(HeaderMode.raw);
properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class);
properties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class);
properties.setPartitionCount(6);
DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties));
Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, producerProperties);
DirectChannel output = createBindableChannel("output", createProducerBindingProperties(properties));
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, properties);
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.setConcurrency(2);
@@ -63,6 +62,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
consumerProperties.setInstanceIndex(0);
consumerProperties.setPartitioned(true);
consumerProperties.setHeaderMode(HeaderMode.raw);
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
QueueChannel input0 = new QueueChannel();
input0.setBeanName("test.input0J");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partJ.0", "test", input0, consumerProperties);
@@ -75,9 +75,9 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
input2.setBeanName("test.input2J");
Binding<MessageChannel> input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties);
output.send(new GenericMessage<>(new byte[] { (byte) 0 }));
output.send(new GenericMessage<>(new byte[] { (byte) 1 }));
output.send(new GenericMessage<>(new byte[] { (byte) 2 }));
output.send(new GenericMessage<>(new byte[] {(byte) 0}));
output.send(new GenericMessage<>(new byte[] {(byte) 1}));
output.send(new GenericMessage<>(new byte[] {(byte) 2}));
Message<?> receive0 = receive(input0);
assertThat(receive0).isNotNull();
@@ -99,26 +99,31 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
@Override
public void testPartitionedModuleSpEL() throws Exception {
KafkaTestBinder binder = getBinder();
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]"));
producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
producerProperties.setPartitionCount(6);
producerProperties.setHeaderMode(HeaderMode.raw);
DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties));
ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]"));
properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
properties.setPartitionCount(6);
properties.setHeaderMode(HeaderMode.raw);
DirectChannel output = createBindableChannel("output", createProducerBindingProperties(properties));
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, producerProperties);
Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, properties);
try {
Lifecycle endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint)).contains("part.0-' + headers['partition']");
Object endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint))
.contains(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']");
}
catch (UnsupportedOperationException ignored) {
}
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.setConcurrency(2);
consumerProperties.setInstanceIndex(0);
consumerProperties.setInstanceCount(3);
consumerProperties.setPartitioned(true);
consumerProperties.setHeaderMode(HeaderMode.raw);
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
QueueChannel input0 = new QueueChannel();
input0.setBeanName("test.input0S");
Binding<MessageChannel> input0Binding = binder.bindConsumer("part.0", "test", input0, consumerProperties);
@@ -131,13 +136,13 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
input2.setBeanName("test.input2S");
Binding<MessageChannel> input2Binding = binder.bindConsumer("part.0", "test", input2, consumerProperties);
Message<byte[]> message2 = MessageBuilder.withPayload(new byte[] { 2 })
Message<byte[]> message2 = MessageBuilder.withPayload(new byte[] {2})
.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo")
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build();
output.send(message2);
output.send(new GenericMessage<>(new byte[] { 1 }));
output.send(new GenericMessage<>(new byte[] { 0 }));
output.send(new GenericMessage<>(new byte[] {1}));
output.send(new GenericMessage<>(new byte[] {0}));
Message<?> receive0 = receive(input0);
assertThat(receive0).isNotNull();
Message<?> receive1 = receive(input1);
@@ -156,15 +161,14 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
@Override
public void testSendAndReceive() throws Exception {
KafkaTestBinder binder = getBinder();
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
DirectChannel moduleOutputChannel = createBindableChannel("output",
createProducerBindingProperties(producerProperties));
producerProperties.setHeaderMode(HeaderMode.raw);
Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel,
producerProperties);
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.setHeaderMode(HeaderMode.raw);
QueueChannel moduleInputChannel = new QueueChannel();
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel,
consumerProperties);
Message<?> message = MessageBuilder.withPayload("foo".getBytes()).build();
@@ -178,14 +182,6 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
consumerBinding.unbind();
}
// Ignored, since raw mode does not support headers
@Test
@Override
@Ignore
public void testSendAndReceiveNoOriginalContentType() throws Exception {
}
@Test
public void testSendAndReceiveWithExplicitConsumerGroup() {
KafkaTestBinder binder = getBinder();
@@ -196,9 +192,11 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
QueueChannel module3InputChannel = new QueueChannel();
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
producerProperties.setHeaderMode(HeaderMode.raw);
Binding<MessageChannel> producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, producerProperties);
Binding<MessageChannel> producerBinding = binder.bindProducer("baz.0", moduleOutputChannel,
producerProperties);
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.setHeaderMode(HeaderMode.raw);
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
Binding<MessageChannel> input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel,
consumerProperties);
// A new module is using the tap as an input channel