Compare commits
29 Commits
v1.3.4.REL
...
0.11
| Author | SHA1 | Date | |
|---|---|---|---|
|
4a5bc655e4 | ||
|
|
6029da2cb2 | ||
|
337596e334 | ||
|
|
6614ab90d3 | ||
|
|
fa6a476996 | ||
|
|
dce3e83520 | ||
|
|
117c628f88 | ||
|
2a3300d066 | ||
|
68fad4a958 | ||
|
11e82813e3 | ||
|
|
b52b398dab | ||
|
|
a79bd5f15f | ||
|
|
f9c62af169 | ||
|
|
4b9ab0448e | ||
|
|
b8c1ca3533 | ||
|
5269a6642c | ||
|
|
a6201425e0 | ||
|
|
715b69506d | ||
|
|
4fb37e2eac | ||
|
|
2a6d716ecf | ||
|
|
7344bed048 | ||
|
|
1e74d624c8 | ||
|
|
7a2b9f1c56 | ||
|
|
d72a4fdb33 | ||
|
|
ea8b3062f1 | ||
|
|
3b3896d642 | ||
|
|
3d244b156e | ||
|
|
e968194b1a | ||
|
|
add9a5d1dc |
39
pom.xml
39
pom.xml
@@ -1,43 +1,42 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-parent</artifactId>
|
||||
<version>1.3.1.BUILD-SNAPSHOT</version>
|
||||
<packaging>pom</packaging>
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-build</artifactId>
|
||||
<version>1.3.12.RELEASE</version>
|
||||
<version>1.3.5.RELEASE</version>
|
||||
<relativePath />
|
||||
</parent>
|
||||
<properties>
|
||||
<java.version>1.7</java.version>
|
||||
<kafka.version>0.10.1.1</kafka.version>
|
||||
<spring-kafka.version>1.1.6.RELEASE</spring-kafka.version>
|
||||
<spring-integration-kafka.version>2.1.2.RELEASE</spring-integration-kafka.version>
|
||||
<spring-cloud-stream.version>1.3.4.RELEASE</spring-cloud-stream.version>
|
||||
<spring-cloud-build.version>1.3.9.RELEASE</spring-cloud-build.version>
|
||||
<kafka.version>0.11.0.0</kafka.version>
|
||||
<spring-kafka.version>1.3.2.RELEASE</spring-kafka.version>
|
||||
<spring-integration-kafka.version>2.3.0.RELEASE</spring-integration-kafka.version>
|
||||
<spring-cloud-stream.version>1.3.1.BUILD-SNAPSHOT</spring-cloud-stream.version>
|
||||
<spring-cloud-build.version>1.3.5.RELEASE</spring-cloud-build.version>
|
||||
</properties>
|
||||
<modules>
|
||||
<module>spring-cloud-stream-binder-kafka</module>
|
||||
<module>spring-cloud-starter-stream-kafka</module>
|
||||
<module>spring-cloud-stream-binder-kafka-docs</module>
|
||||
<module>spring-cloud-stream-binder-kafka-0.10.1-test</module>
|
||||
<module>spring-cloud-stream-binder-kafka-0.10.2-test</module>
|
||||
<module>spring-cloud-stream-binder-kafka-core</module>
|
||||
<module>spring-cloud-stream-binder-kstream</module>
|
||||
<module>spring-cloud-stream-binder-kafka11</module>
|
||||
<module>spring-cloud-starter-stream-kafka11</module>
|
||||
<module>spring-cloud-stream-binder-kafka11-docs</module>
|
||||
<module>spring-cloud-stream-binder-kafka-0.11-test</module>
|
||||
<module>spring-cloud-stream-binder-kafka11-core</module>
|
||||
<module>spring-cloud-stream-binder-kstream11</module>
|
||||
</modules>
|
||||
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
@@ -83,6 +82,12 @@
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-kafka</artifactId>
|
||||
<version>${spring-integration-kafka.version}</version>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<groupId>org.apache.avro</groupId>
|
||||
<artifactId>avro-compiler</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
|
||||
@@ -3,11 +3,11 @@
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-parent</artifactId>
|
||||
<version>1.3.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
|
||||
<description>Spring Cloud Starter Stream Kafka</description>
|
||||
<artifactId>spring-cloud-starter-stream-kafka11</artifactId>
|
||||
<description>Spring Cloud Starter Stream Kafka for the 0.11.x.x client</description>
|
||||
<url>https://projects.spring.io/spring-cloud</url>
|
||||
<organization>
|
||||
<name>Pivotal Software, Inc.</name>
|
||||
@@ -19,7 +19,7 @@
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
@@ -1,132 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
</parent>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-0.10.1-test</artifactId>
|
||||
<description>Spring Cloud Stream Kafka Binder 0.10.1 Tests</description>
|
||||
<url>https://projects.spring.io/spring-cloud</url>
|
||||
<organization>
|
||||
<name>Pivotal Software, Inc.</name>
|
||||
<url>https://www.spring.io</url>
|
||||
</organization>
|
||||
<properties>
|
||||
<main.basedir>${basedir}/../..</main.basedir>
|
||||
<kafka.version>0.10.1.1</kafka.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka_2.11</artifactId>
|
||||
<scope>test</scope>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-log4j12</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka-clients</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-kafka</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-jmx</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
<version>${project.version}</version>
|
||||
<type>test-jar</type>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-0.10.2-test</artifactId>
|
||||
<version>${project.version}</version>
|
||||
<type>test-jar</type>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-schema</artifactId>
|
||||
<version>${spring-cloud-stream.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.confluent</groupId>
|
||||
<artifactId>kafka-avro-serializer</artifactId>
|
||||
<version>3.1.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.confluent</groupId>
|
||||
<artifactId>kafka-schema-registry</artifactId>
|
||||
<version>3.1.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<repositories>
|
||||
<repository>
|
||||
<id>confluent</id>
|
||||
<url>https://packages.confluent.io/maven/</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-jar-plugin</artifactId>
|
||||
<version>3.0.2</version>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>test-jar</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@@ -1,244 +0,0 @@
|
||||
/*
|
||||
* Copyright 2014-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
|
||||
*
|
||||
* https://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 io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
|
||||
import io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication;
|
||||
import kafka.utils.ZKStringSerializer$;
|
||||
import kafka.utils.ZkUtils;
|
||||
import org.I0Itec.zkclient.ZkClient;
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
|
||||
import org.apache.kafka.common.serialization.Deserializer;
|
||||
import org.assertj.core.api.Assertions;
|
||||
import org.eclipse.jetty.server.Server;
|
||||
import org.junit.Before;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
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.Spy;
|
||||
import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.kafka.core.ConsumerFactory;
|
||||
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
|
||||
import org.springframework.kafka.support.KafkaHeaders;
|
||||
import org.springframework.kafka.test.core.BrokerAddress;
|
||||
import org.springframework.kafka.test.rule.KafkaEmbedded;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
import java.util.UUID;
|
||||
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
/**
|
||||
* Integration tests for the {@link KafkaMessageChannelBinder}.
|
||||
*
|
||||
* This test specifically tests for the 0.10.1.x version of Kafka.
|
||||
*
|
||||
* @author Eric Bottard
|
||||
* @author Marius Bogoevici
|
||||
* @author Mark Fisher
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*/
|
||||
public class Kafka_0_10_1_BinderTests extends Kafka_0_10_2_BinderTests {
|
||||
|
||||
private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName();
|
||||
|
||||
@ClassRule
|
||||
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10);
|
||||
|
||||
private Kafka10TestBinder binder;
|
||||
|
||||
private final Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation();
|
||||
|
||||
@Override
|
||||
protected void binderBindUnbindLatency() throws InterruptedException {
|
||||
Thread.sleep(500);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Kafka10TestBinder getBinder() {
|
||||
if (binder == null) {
|
||||
KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
|
||||
binderConfiguration.setHeaders("dlqTestHeader");
|
||||
binder = new Kafka10TestBinder(binderConfiguration);
|
||||
}
|
||||
return binder;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected KafkaBinderConfigurationProperties createConfigurationProperties() {
|
||||
KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties();
|
||||
BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();
|
||||
List<String> bAddresses = new ArrayList<>();
|
||||
for (BrokerAddress bAddress : brokerAddresses) {
|
||||
bAddresses.add(bAddress.toString());
|
||||
}
|
||||
String[] foo = new String[bAddresses.size()];
|
||||
binderConfiguration.setBrokers(bAddresses.toArray(foo));
|
||||
binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString());
|
||||
return binderConfiguration;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected int partitionSize(String topic) {
|
||||
return consumerFactory().createConsumer().partitionsFor(topic).size();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) {
|
||||
final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(),
|
||||
kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(),
|
||||
ZKStringSerializer$.MODULE$);
|
||||
|
||||
return new ZkUtils(zkClient, null, false);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) {
|
||||
adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected int invokePartitionSize(String topic, ZkUtils zkUtils) {
|
||||
return adminUtilsOperation.partitionSize(topic, zkUtils);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getKafkaOffsetHeaderKey() {
|
||||
return KafkaHeaders.OFFSET;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) {
|
||||
return new Kafka10TestBinder(kafkaBinderConfigurationProperties);
|
||||
}
|
||||
|
||||
@Before
|
||||
public void init() {
|
||||
String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER");
|
||||
if (multiplier != null) {
|
||||
timeoutMultiplier = Double.parseDouble(multiplier);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean usesExplicitRouting() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getClassUnderTestName() {
|
||||
return CLASS_UNDER_TEST_NAME;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Spy spyOn(final String name) {
|
||||
throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests");
|
||||
}
|
||||
|
||||
|
||||
private ConsumerFactory<byte[], byte[]> consumerFactory() {
|
||||
Map<String, Object> props = new HashMap<>();
|
||||
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
|
||||
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString());
|
||||
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
|
||||
props.put(ConsumerConfig.GROUP_ID_CONFIG, "TEST-CONSUMER-GROUP");
|
||||
Deserializer<byte[]> valueDecoder = new ByteArrayDeserializer();
|
||||
Deserializer<byte[]> keyDecoder = new ByteArrayDeserializer();
|
||||
|
||||
return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder);
|
||||
}
|
||||
|
||||
@Ignore
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testCustomAvroSerialization() throws Exception {
|
||||
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
|
||||
final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
|
||||
configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(),
|
||||
ZKStringSerializer$.MODULE$);
|
||||
final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
|
||||
Map<String, Object> schemaRegistryProps = new HashMap<>();
|
||||
schemaRegistryProps.put("kafkastore.connection.url", configurationProperties.getZkConnectionString());
|
||||
schemaRegistryProps.put("listeners", "https://0.0.0.0:8082");
|
||||
schemaRegistryProps.put("port", "8082");
|
||||
schemaRegistryProps.put("kafkastore.topic", "_schemas");
|
||||
SchemaRegistryConfig config = new SchemaRegistryConfig(schemaRegistryProps);
|
||||
SchemaRegistryRestApplication app = new SchemaRegistryRestApplication(config);
|
||||
Server server = app.createServer();
|
||||
server.start();
|
||||
long endTime = System.currentTimeMillis() + 5000;
|
||||
while(true) {
|
||||
if (server.isRunning()) {
|
||||
break;
|
||||
}
|
||||
else if (System.currentTimeMillis() > endTime) {
|
||||
Assertions.fail("Kafka Schema Registry Server failed to start");
|
||||
}
|
||||
}
|
||||
User1 firstOutboundFoo = new User1();
|
||||
String userName1 = "foo-name" + UUID.randomUUID().toString();
|
||||
String favColor1 = "foo-color" + UUID.randomUUID().toString();
|
||||
firstOutboundFoo.setName(userName1);
|
||||
firstOutboundFoo.setFavoriteColor(favColor1);
|
||||
Message<?> message = MessageBuilder.withPayload(firstOutboundFoo).build();
|
||||
SubscribableChannel moduleOutputChannel = new DirectChannel();
|
||||
String testTopicName = "existing" + System.currentTimeMillis();
|
||||
invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties());
|
||||
configurationProperties.setAutoAddPartitions(true);
|
||||
Binder binder = getBinder(configurationProperties);
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
|
||||
producerProperties.getExtension().getConfiguration().put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
|
||||
producerProperties.getExtension().getConfiguration().put("schema.registry.url", "http://localhost:8082");
|
||||
producerProperties.setUseNativeEncoding(true);
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties);
|
||||
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
|
||||
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
|
||||
consumerProperties.getExtension().getConfiguration().put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
|
||||
consumerProperties.getExtension().getConfiguration().put("schema.registry.url", "http://localhost:8082");
|
||||
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties);
|
||||
// Let the consumer actually bind to the producer before sending a msg
|
||||
binderBindUnbindLatency();
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = receive(moduleInputChannel);
|
||||
Assertions.assertThat(inbound).isNotNull();
|
||||
assertTrue(message.getPayload() instanceof User1);
|
||||
User1 receivedUser = (User1) message.getPayload();
|
||||
Assertions.assertThat(receivedUser.getName()).isEqualTo(userName1);
|
||||
Assertions.assertThat(receivedUser.getFavoriteColor()).isEqualTo(favColor1);
|
||||
producerBinding.unbind();
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
}
|
||||
@@ -3,11 +3,11 @@
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-parent</artifactId>
|
||||
<version>1.3.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-0.10.2-test</artifactId>
|
||||
<description>Spring Cloud Stream Kafka Binder 0.10.2 Tests</description>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-0.11-test</artifactId>
|
||||
<description>Spring Cloud Stream Kafka Binder 0.11 Tests</description>
|
||||
<url>https://projects.spring.io/spring-cloud</url>
|
||||
<organization>
|
||||
<name>Pivotal Software, Inc.</name>
|
||||
@@ -15,18 +15,18 @@
|
||||
</organization>
|
||||
<properties>
|
||||
<main.basedir>${basedir}/../..</main.basedir>
|
||||
<kafka.version>0.10.2.1</kafka.version>
|
||||
<spring-kafka.version>1.2.2.RELEASE</spring-kafka.version>
|
||||
<kafka.version>0.11.0.0</kafka.version>
|
||||
<spring-kafka.version>1.3.0.RELEASE</spring-kafka.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
@@ -59,17 +59,9 @@
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-kafka</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-jmx</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11</artifactId>
|
||||
<version>${project.version}</version>
|
||||
<type>test-jar</type>
|
||||
<scope>test</scope>
|
||||
@@ -37,10 +37,10 @@ import org.assertj.core.api.Assertions;
|
||||
import org.eclipse.jetty.server.Server;
|
||||
import org.junit.Before;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.cloud.stream.binder.BinderHeaders;
|
||||
import org.springframework.cloud.stream.binder.Binding;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
|
||||
@@ -49,6 +49,7 @@ import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOper
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
|
||||
import org.springframework.cloud.stream.config.BindingProperties;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.kafka.core.ConsumerFactory;
|
||||
@@ -58,15 +59,18 @@ import org.springframework.kafka.test.core.BrokerAddress;
|
||||
import org.springframework.kafka.test.rule.KafkaEmbedded;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.util.MimeType;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
/**
|
||||
* Integration tests for the {@link KafkaMessageChannelBinder}.
|
||||
*
|
||||
* This test specifically tests for the 0.10.2.x version of Kafka.
|
||||
* This test specifically tests for the 0.11.x.x version of Kafka.
|
||||
*
|
||||
* @author Eric Bottard
|
||||
* @author Marius Bogoevici
|
||||
@@ -74,7 +78,7 @@ import static org.junit.Assert.assertTrue;
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class Kafka_0_10_2_BinderTests extends KafkaBinderTests {
|
||||
public class Kafka_0_11_BinderTests extends KafkaBinderTests {
|
||||
|
||||
private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName();
|
||||
|
||||
@@ -184,7 +188,42 @@ public class Kafka_0_10_2_BinderTests extends KafkaBinderTests {
|
||||
return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder);
|
||||
}
|
||||
|
||||
@Ignore
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void testTrustedPackages() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
|
||||
BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties());
|
||||
DirectChannel moduleOutputChannel = createBindableChannel("output", producerBindingProperties);
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer("bar.0", moduleOutputChannel,
|
||||
producerBindingProperties.getProducer());
|
||||
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
|
||||
consumerProperties.getExtension().setTrustedPackages(new String[]{"org.springframework.util"});
|
||||
Binding<MessageChannel> consumerBinding = binder.bindConsumer("bar.0",
|
||||
"testSendAndReceiveNoOriginalContentType", moduleInputChannel, consumerProperties);
|
||||
binderBindUnbindLatency();
|
||||
|
||||
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("foo")
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).setHeader("foo", MimeTypeUtils.TEXT_PLAIN)
|
||||
.build();
|
||||
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = receive(moduleInputChannel);
|
||||
Assertions.assertThat(inbound).isNotNull();
|
||||
Assertions.assertThat(inbound.getPayload()).isEqualTo("foo");
|
||||
Assertions.assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
Assertions.assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE))
|
||||
.isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
|
||||
Assertions.assertThat(inbound.getHeaders().get("foo")).isInstanceOf(MimeType.class);
|
||||
MimeType actual = (MimeType) inbound.getHeaders().get("foo");
|
||||
Assertions.assertThat(actual).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
|
||||
producerBinding.unbind();
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
|
||||
class Foo{}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testCustomAvroSerialization() throws Exception {
|
||||
@@ -245,4 +284,15 @@ public class Kafka_0_10_2_BinderTests extends KafkaBinderTests {
|
||||
producerBinding.unbind();
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void testSendAndReceiveWithExplicitConsumerGroupWithRawMode() {
|
||||
// raw mode no longer needed
|
||||
}
|
||||
|
||||
@Override
|
||||
public void testSendAndReceiveWithRawModeAndStringPayload() {
|
||||
// raw mode no longer needed
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,118 +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
|
||||
*
|
||||
* https://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.File;
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import javax.security.auth.login.AppConfigurationEntry;
|
||||
import javax.security.auth.login.Configuration;
|
||||
|
||||
import org.apache.kafka.common.security.JaasUtils;
|
||||
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.ApplicationContextAware;
|
||||
import org.springframework.context.ApplicationListener;
|
||||
import org.springframework.context.event.ContextRefreshedEvent;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class KafkaBinderJaasInitializerListener implements ApplicationListener<ContextRefreshedEvent>,
|
||||
ApplicationContextAware, DisposableBean {
|
||||
|
||||
public static final String DEFAULT_ZK_LOGIN_CONTEXT_NAME = "Client";
|
||||
|
||||
private ApplicationContext applicationContext;
|
||||
|
||||
private final boolean ignoreJavaLoginConfigParamSystemProperty;
|
||||
|
||||
private final File placeholderJaasConfiguration;
|
||||
|
||||
public KafkaBinderJaasInitializerListener() throws IOException {
|
||||
// we ignore the system property if it wasn't originally set at launch
|
||||
this.ignoreJavaLoginConfigParamSystemProperty =
|
||||
(System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM) == null);
|
||||
this.placeholderJaasConfiguration = File.createTempFile("kafka-client-jaas-config-placeholder", "conf");
|
||||
this.placeholderJaasConfiguration.deleteOnExit();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
|
||||
this.applicationContext = applicationContext;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() throws Exception {
|
||||
if (this.ignoreJavaLoginConfigParamSystemProperty) {
|
||||
System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onApplicationEvent(ContextRefreshedEvent event) {
|
||||
if (event.getSource() == this.applicationContext) {
|
||||
KafkaBinderConfigurationProperties binderConfigurationProperties =
|
||||
applicationContext.getBean(KafkaBinderConfigurationProperties.class);
|
||||
// only use programmatic support if a file is not set via system property
|
||||
if (ignoreJavaLoginConfigParamSystemProperty
|
||||
&& binderConfigurationProperties.getJaas() != null) {
|
||||
Map<String, AppConfigurationEntry[]> configurationEntries = new HashMap<>();
|
||||
AppConfigurationEntry kafkaClientConfigurationEntry = new AppConfigurationEntry
|
||||
(binderConfigurationProperties.getJaas().getLoginModule(),
|
||||
binderConfigurationProperties.getJaas().getControlFlagValue(),
|
||||
binderConfigurationProperties.getJaas().getOptions() != null ?
|
||||
binderConfigurationProperties.getJaas().getOptions() :
|
||||
Collections.<String, Object>emptyMap());
|
||||
configurationEntries.put(JaasUtils.LOGIN_CONTEXT_CLIENT,
|
||||
new AppConfigurationEntry[]{ kafkaClientConfigurationEntry });
|
||||
Configuration.setConfiguration(new InternalConfiguration(configurationEntries));
|
||||
// Workaround for a 0.9 client issue where even if the Configuration is set
|
||||
// a system property check is performed.
|
||||
// Since the Configuration already exists, this will be ignored.
|
||||
if (this.placeholderJaasConfiguration != null) {
|
||||
System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, this.placeholderJaasConfiguration.getAbsolutePath());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@link Configuration} set up programmatically by the Kafka binder
|
||||
*/
|
||||
public static class InternalConfiguration extends Configuration {
|
||||
|
||||
private final Map<String, AppConfigurationEntry[]> configurationEntries;
|
||||
|
||||
public InternalConfiguration(Map<String, AppConfigurationEntry[]> configurationEntries) {
|
||||
Assert.notNull(configurationEntries, " cannot be null");
|
||||
Assert.notEmpty(configurationEntries, " cannot be empty");
|
||||
this.configurationEntries = configurationEntries;
|
||||
}
|
||||
|
||||
@Override
|
||||
public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
|
||||
return configurationEntries.get(name);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,61 +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
|
||||
*
|
||||
* https://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 javax.security.auth.login.AppConfigurationEntry;
|
||||
|
||||
import com.sun.security.auth.login.ConfigFile;
|
||||
|
||||
import org.apache.kafka.common.security.JaasUtils;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class KafkaBinderJaasInitializerListenerTest {
|
||||
|
||||
@Test
|
||||
public void testConfigurationParsedCorrectlyWithKafkaClient() throws Exception {
|
||||
ConfigFile configFile = new ConfigFile(new ClassPathResource("jaas-sample-kafka-only.conf").getURI());
|
||||
final AppConfigurationEntry[] kafkaConfigurationArray = configFile.getAppConfigurationEntry(JaasUtils.LOGIN_CONTEXT_CLIENT);
|
||||
|
||||
final ConfigurableApplicationContext context =
|
||||
SpringApplication.run(SimpleApplication.class,
|
||||
"--spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true",
|
||||
"--spring.cloud.stream.kafka.binder.jaas.options.storeKey=true",
|
||||
"--spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab",
|
||||
"--spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM");
|
||||
javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration.getConfiguration();
|
||||
|
||||
final AppConfigurationEntry[] kafkaConfiguration = configuration.getAppConfigurationEntry(JaasUtils.LOGIN_CONTEXT_CLIENT);
|
||||
assertThat(kafkaConfiguration).hasSize(1);
|
||||
assertThat(kafkaConfiguration[0].getOptions()).isEqualTo(kafkaConfigurationArray[0].getOptions());
|
||||
context.close();
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class SimpleApplication {
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
/*
|
||||
* Copyright 2018 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
|
||||
*
|
||||
* https://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.Collections;
|
||||
|
||||
import org.apache.kafka.clients.producer.Callback;
|
||||
import org.apache.kafka.clients.producer.Producer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.apache.kafka.common.TopicPartition;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.mockito.InOrder;
|
||||
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
|
||||
import org.springframework.kafka.test.rule.KafkaEmbedded;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.BDDMockito.willReturn;
|
||||
import static org.mockito.Mockito.inOrder;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.spy;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*
|
||||
*/
|
||||
public class KafkaTransactionTests {
|
||||
|
||||
@ClassRule
|
||||
public static final KafkaEmbedded embeddedKafka = new KafkaEmbedded(1);
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void testProducerRunsInTx() {
|
||||
KafkaProperties kafkaProperties = new KafkaProperties();
|
||||
kafkaProperties.setBootstrapServers(Collections.singletonList(embeddedKafka.getBrokersAsString()));
|
||||
KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties();
|
||||
configurationProperties.getTransaction().setTransactionIdPrefix("foo-");
|
||||
KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(configurationProperties, kafkaProperties);
|
||||
provisioningProvider.setMetadataRetryOperations(new RetryTemplate());
|
||||
final Producer mockProducer = mock(Producer.class);
|
||||
willReturn(Collections.singletonList(new TopicPartition("foo", 0))).given(mockProducer).partitionsFor(anyString());
|
||||
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties, provisioningProvider) {
|
||||
|
||||
@Override
|
||||
protected DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(String transactionIdPrefix,
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
|
||||
DefaultKafkaProducerFactory<byte[], byte[]> producerFactory =
|
||||
spy(super.getProducerFactory(transactionIdPrefix, producerProperties));
|
||||
willReturn(mockProducer).given(producerFactory).createProducer();
|
||||
return producerFactory;
|
||||
}
|
||||
|
||||
};
|
||||
GenericApplicationContext applicationContext = new GenericApplicationContext();
|
||||
applicationContext.refresh();
|
||||
binder.setApplicationContext(applicationContext);
|
||||
DirectChannel channel = new DirectChannel();
|
||||
KafkaProducerProperties extension = new KafkaProducerProperties();
|
||||
ExtendedProducerProperties<KafkaProducerProperties> properties = new ExtendedProducerProperties<>(extension);
|
||||
binder.bindProducer("foo", channel, properties);
|
||||
channel.send(new GenericMessage<>("foo".getBytes()));
|
||||
InOrder inOrder = inOrder(mockProducer);
|
||||
inOrder.verify(mockProducer).beginTransaction();
|
||||
inOrder.verify(mockProducer).send(any(ProducerRecord.class), any(Callback.class));
|
||||
inOrder.verify(mockProducer).commitTransaction();
|
||||
inOrder.verify(mockProducer).close();
|
||||
inOrder.verifyNoMoreInteractions();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -3,11 +3,11 @@
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-parent</artifactId>
|
||||
<version>1.3.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
|
||||
<description>Spring Cloud Stream Kafka Binder Core</description>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-core</artifactId>
|
||||
<description>Spring Cloud Stream Kafka Binder Core for the 0.11.x.x client</description>
|
||||
<url>https://projects.spring.io/spring-cloud</url>
|
||||
<organization>
|
||||
<name>Pivotal Software, Inc.</name>
|
||||
@@ -33,13 +33,6 @@
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-kafka</artifactId>
|
||||
<version>${spring-integration-kafka.version}</version>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<groupId>org.apache.avro</groupId>
|
||||
<artifactId>avro-compiler</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
@@ -27,6 +27,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
@@ -40,6 +41,8 @@ import org.springframework.util.StringUtils;
|
||||
@ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder")
|
||||
public class KafkaBinderConfigurationProperties {
|
||||
|
||||
private final Transaction transaction = new Transaction();
|
||||
|
||||
@Autowired(required = false)
|
||||
private KafkaProperties kafkaProperties;
|
||||
|
||||
@@ -96,6 +99,15 @@ public class KafkaBinderConfigurationProperties {
|
||||
|
||||
private JaasLoginModuleConfiguration jaas;
|
||||
|
||||
/**
|
||||
* The bean name of a custom header mapper to use instead of a {@link DefaultKafkaHeaderMapper}.
|
||||
*/
|
||||
private String headerMapperBeanName;
|
||||
|
||||
public Transaction getTransaction() {
|
||||
return this.transaction;
|
||||
}
|
||||
|
||||
public String getZkConnectionString() {
|
||||
return toConnectionString(this.zkNodes, this.defaultZkPort);
|
||||
}
|
||||
@@ -352,4 +364,32 @@ public class KafkaBinderConfigurationProperties {
|
||||
this.jaas = jaas;
|
||||
}
|
||||
|
||||
public String getHeaderMapperBeanName() {
|
||||
return this.headerMapperBeanName;
|
||||
}
|
||||
|
||||
public void setHeaderMapperBeanName(String headerMapperBeanName) {
|
||||
this.headerMapperBeanName = headerMapperBeanName;
|
||||
}
|
||||
|
||||
public static class Transaction {
|
||||
|
||||
private final KafkaProducerProperties producer = new KafkaProducerProperties();
|
||||
|
||||
private String transactionIdPrefix;
|
||||
|
||||
public String getTransactionIdPrefix() {
|
||||
return this.transactionIdPrefix;
|
||||
}
|
||||
|
||||
public void setTransactionIdPrefix(String transactionIdPrefix) {
|
||||
this.transactionIdPrefix = transactionIdPrefix;
|
||||
}
|
||||
|
||||
public KafkaProducerProperties getProducer() {
|
||||
return this.producer;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -22,6 +22,8 @@ import java.util.Map;
|
||||
/**
|
||||
* @author Marius Bogoevici
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author Soby Chacko
|
||||
* @author Gary Russell
|
||||
*
|
||||
* <p>
|
||||
* Thanks to Laszlo Szabo for providing the initial patch for generic property support.
|
||||
@@ -29,6 +31,28 @@ import java.util.Map;
|
||||
*/
|
||||
public class KafkaConsumerProperties {
|
||||
|
||||
public enum StartOffset {
|
||||
earliest(-2L),
|
||||
latest(-1L);
|
||||
|
||||
private final long referencePoint;
|
||||
|
||||
StartOffset(long referencePoint) {
|
||||
this.referencePoint = referencePoint;
|
||||
}
|
||||
|
||||
public long getReferencePoint() {
|
||||
return this.referencePoint;
|
||||
}
|
||||
}
|
||||
|
||||
public enum StandardHeaders {
|
||||
none,
|
||||
id,
|
||||
timestamp,
|
||||
both
|
||||
}
|
||||
|
||||
private boolean autoRebalanceEnabled = true;
|
||||
|
||||
private boolean autoCommitOffset = true;
|
||||
@@ -43,6 +67,12 @@ public class KafkaConsumerProperties {
|
||||
|
||||
private int recoveryInterval = 5000;
|
||||
|
||||
private String[] trustedPackages;
|
||||
|
||||
private StandardHeaders standardHeaders = StandardHeaders.none;
|
||||
|
||||
private String converterBeanName;
|
||||
|
||||
private Map<String, String> configuration = new HashMap<>();
|
||||
|
||||
public boolean isAutoCommitOffset() {
|
||||
@@ -93,21 +123,6 @@ public class KafkaConsumerProperties {
|
||||
this.autoRebalanceEnabled = autoRebalanceEnabled;
|
||||
}
|
||||
|
||||
public enum StartOffset {
|
||||
earliest(-2L),
|
||||
latest(-1L);
|
||||
|
||||
private final long referencePoint;
|
||||
|
||||
StartOffset(long referencePoint) {
|
||||
this.referencePoint = referencePoint;
|
||||
}
|
||||
|
||||
public long getReferencePoint() {
|
||||
return this.referencePoint;
|
||||
}
|
||||
}
|
||||
|
||||
public Map<String, String> getConfiguration() {
|
||||
return this.configuration;
|
||||
}
|
||||
@@ -123,4 +138,29 @@ public class KafkaConsumerProperties {
|
||||
public void setDlqName(String dlqName) {
|
||||
this.dlqName = dlqName;
|
||||
}
|
||||
|
||||
public String[] getTrustedPackages() {
|
||||
return trustedPackages;
|
||||
}
|
||||
|
||||
public void setTrustedPackages(String[] trustedPackages) {
|
||||
this.trustedPackages = trustedPackages;
|
||||
}
|
||||
|
||||
public StandardHeaders getStandardHeaders() {
|
||||
return this.standardHeaders;
|
||||
}
|
||||
|
||||
public void setStandardHeaders(StandardHeaders standardHeaders) {
|
||||
this.standardHeaders = standardHeaders;
|
||||
}
|
||||
|
||||
public String getConverterBeanName() {
|
||||
return this.converterBeanName;
|
||||
}
|
||||
|
||||
public void setConverterBeanName(String converterBeanName) {
|
||||
this.converterBeanName = converterBeanName;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -16,16 +16,17 @@
|
||||
|
||||
package org.springframework.cloud.stream.binder.kafka.properties;
|
||||
|
||||
import org.springframework.expression.Expression;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
import org.springframework.expression.Expression;
|
||||
|
||||
/**
|
||||
* @author Marius Bogoevici
|
||||
* @author Henryk Konsek
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class KafkaProducerProperties {
|
||||
|
||||
@@ -39,6 +40,8 @@ public class KafkaProducerProperties {
|
||||
|
||||
private Expression messageKeyExpression;
|
||||
|
||||
private String[] headerPatterns;
|
||||
|
||||
private Map<String, String> configuration = new HashMap<>();
|
||||
|
||||
public int getBufferSize() {
|
||||
@@ -82,6 +85,14 @@ public class KafkaProducerProperties {
|
||||
this.messageKeyExpression = messageKeyExpression;
|
||||
}
|
||||
|
||||
public String[] getHeaderPatterns() {
|
||||
return this.headerPatterns;
|
||||
}
|
||||
|
||||
public void setHeaderPatterns(String[] headerPatterns) {
|
||||
this.headerPatterns = headerPatterns;
|
||||
}
|
||||
|
||||
public Map<String, String> getConfiguration() {
|
||||
return this.configuration;
|
||||
}
|
||||
@@ -4,13 +4,13 @@
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-parent</artifactId>
|
||||
<version>1.3.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>spring-cloud-stream-binder-kafka-docs</artifactId>
|
||||
<name>spring-cloud-stream-binder-kafka-docs</name>
|
||||
<description>Spring Cloud Stream Kafka Binder Docs</description>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-docs</artifactId>
|
||||
<name>spring-cloud-stream-binder-kafka11-docs</name>
|
||||
<description>Spring Cloud Stream Kafka Binder Docs for the 0.11.x.x client</description>
|
||||
<properties>
|
||||
<main.basedir>${basedir}/..</main.basedir>
|
||||
</properties>
|
||||
|
Before Width: | Height: | Size: 9.2 KiB After Width: | Height: | Size: 9.2 KiB |
@@ -26,6 +26,8 @@ include::overview.adoc[]
|
||||
|
||||
include::dlq.adoc[]
|
||||
|
||||
include::metrics.adoc[]
|
||||
|
||||
= Appendices
|
||||
[appendix]
|
||||
include::building.adoc[]
|
||||
@@ -0,0 +1,10 @@
|
||||
[[kafka-metrics]]
|
||||
== Kafka metrics
|
||||
|
||||
Kafka binder module exposes the following metrics:
|
||||
|
||||
`spring.cloud.stream.binder.kafka.someGroup.someTopic.lag` - this metric indicates how many messages
|
||||
have not been yet consumed from given binder's topic (`someTopic`) by given consumer group (`someGroup`).
|
||||
For example if the value of the metric `spring.cloud.stream.binder.kafka.myGroup.myTopic.lag` is `1000`, then
|
||||
consumer group `myGroup` has `1000` messages to waiting to be consumed from topic `myTopic`. This metric is
|
||||
particularly useful to provide auto-scaling feedback to PaaS platform of your choice.
|
||||
@@ -173,6 +173,16 @@ dlqName::
|
||||
The name of the DLQ topic to receive the error messages.
|
||||
+
|
||||
Default: null (If not specified, messages that result in errors will be forwarded to a topic named `error.<destination>.<group>`).
|
||||
standardHeaders::
|
||||
Indicates which standard headers are populated by the inbound channel adapter.
|
||||
`none`, `id`, `timestamp` or `both`.
|
||||
Useful if using native deserialization and the first component to receive a message needs an `id` (such as an aggregator that is configured to use a JDBC message store).
|
||||
+
|
||||
Default: `none`
|
||||
converterBeanName::
|
||||
The name of a bean that implements `RecordMessageConverter`; used in the inbound channel adapter to replace the default `MessagingMessageConverter`.
|
||||
+
|
||||
Default: `null`
|
||||
|
||||
=== Kafka Producer Properties
|
||||
|
||||
@@ -528,34 +538,6 @@ spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.c
|
||||
spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde
|
||||
----
|
||||
|
||||
timewindow.length::
|
||||
Many streaming applications written using Kafka Streams involve windowning operations.
|
||||
If you specify this property, there is a `org.apache.kafka.streams.kstream.TimeWindows` bean automatically provided that can be autowired in applications.
|
||||
This property must be prefixed with `spring.cloud.stream.kstream.`.
|
||||
A bean of type `org.apache.kafka.streams.kstream.TimeWindows` is created only if this property is provided.
|
||||
|
||||
Following is an example of using this property.
|
||||
Values are provided in milliseconds.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kstream.timeWindow.length=5000
|
||||
----
|
||||
|
||||
timewindow.advanceBy::
|
||||
This property goes hand in hand with `timewindow.length` and has no effect on its own.
|
||||
If you provide this property, the generated `org.apache.kafka.streams.kstream.TimeWindows` bean will automatically conatin this information.
|
||||
This property must be prefixed with `spring.cloud.stream.kstream.`.
|
||||
|
||||
Following is an example of using this property.
|
||||
Values are provided in milliseconds.
|
||||
|
||||
[source]
|
||||
----
|
||||
spring.cloud.stream.kstream.timeWindow.advanceBy=1000
|
||||
----
|
||||
|
||||
|
||||
[[kafka-error-channels]]
|
||||
== Error Channels
|
||||
|
||||
@@ -568,13 +550,3 @@ The payload of the `ErrorMessage` for a send failure is a `KafkaSendFailureExcep
|
||||
* `record` - the raw `ProducerRecord` that was created from the `failedMessage`
|
||||
|
||||
There is no automatic handling of these exceptions (such as sending to a <<kafka-dlq-processing, Dead-Letter queue>>); you can consume these exceptions with your own Spring Integration flow.
|
||||
|
||||
|
||||
[[kafka-metrics]]
|
||||
== Kafka Metrics
|
||||
|
||||
Kafka binder module exposes the following metrics:
|
||||
|
||||
`spring.cloud.stream.binder.kafka.someGroup.someTopic.lag` - this metric indicates how many messages have not been yet consumed from given binder's topic by given consumer group.
|
||||
For example if the value of the metric `spring.cloud.stream.binder.kafka.myGroup.myTopic.lag` is `1000`, then consumer group `myGroup` has `1000` messages to waiting to be consumed from topic `myTopic`.
|
||||
This metric is particularly useful to provide auto-scaling feedback to PaaS platform of your choice.
|
||||
|
Before Width: | Height: | Size: 11 KiB After Width: | Height: | Size: 11 KiB |
|
Before Width: | Height: | Size: 2.0 KiB After Width: | Height: | Size: 2.0 KiB |
|
Before Width: | Height: | Size: 2.0 KiB After Width: | Height: | Size: 2.0 KiB |
|
Before Width: | Height: | Size: 45 KiB After Width: | Height: | Size: 45 KiB |
|
Before Width: | Height: | Size: 2.2 KiB After Width: | Height: | Size: 2.2 KiB |
|
Before Width: | Height: | Size: 931 B After Width: | Height: | Size: 931 B |
|
Before Width: | Height: | Size: 2.1 KiB After Width: | Height: | Size: 2.1 KiB |
@@ -2,21 +2,21 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>spring-cloud-stream-binder-kafka</name>
|
||||
<description>Kafka binder implementation</description>
|
||||
<name>spring-cloud-stream-binder-kafka11</name>
|
||||
<description>Kafka binder implementation for the 0.11.x.x client</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-parent</artifactId>
|
||||
<version>1.3.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
@@ -52,6 +52,7 @@
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka</artifactId>
|
||||
<version>${spring-kafka.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
@@ -61,6 +62,7 @@
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka-test</artifactId>
|
||||
<version>${spring-kafka.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
@@ -19,10 +19,12 @@ package org.springframework.cloud.stream.binder.kafka;
|
||||
import java.io.PrintWriter;
|
||||
import java.io.StringWriter;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
@@ -33,22 +35,25 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
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.common.PartitionInfo;
|
||||
import org.apache.kafka.common.header.Headers;
|
||||
import org.apache.kafka.common.header.internals.RecordHeader;
|
||||
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
|
||||
import org.apache.kafka.common.serialization.ByteArraySerializer;
|
||||
import org.apache.kafka.common.utils.Utils;
|
||||
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
|
||||
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.cloud.stream.binder.BinderHeaders;
|
||||
import org.springframework.cloud.stream.binder.EmbeddedHeaderUtils;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
|
||||
import org.springframework.cloud.stream.binder.HeaderMode;
|
||||
import org.springframework.cloud.stream.binder.MessageValues;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
|
||||
@@ -62,21 +67,31 @@ import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAd
|
||||
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
|
||||
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
|
||||
import org.springframework.integration.support.ErrorMessageStrategy;
|
||||
import org.springframework.kafka.KafkaException;
|
||||
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.AbstractMessageListenerContainer;
|
||||
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
|
||||
import org.springframework.kafka.listener.config.ContainerProperties;
|
||||
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
|
||||
import org.springframework.kafka.support.KafkaHeaderMapper;
|
||||
import org.springframework.kafka.support.ProducerListener;
|
||||
import org.springframework.kafka.support.SendResult;
|
||||
import org.springframework.kafka.support.TopicPartitionInitialOffset;
|
||||
import org.springframework.kafka.support.converter.MessagingMessageConverter;
|
||||
import org.springframework.kafka.transaction.KafkaTransactionManager;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
import org.springframework.transaction.support.TransactionCallback;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
@@ -113,30 +128,28 @@ public class KafkaMessageChannelBinder extends
|
||||
|
||||
private final Map<String, TopicInformation> topicsInUse = new HashMap<>();
|
||||
|
||||
private final KafkaTransactionManager<byte[], byte[]> transactionManager;
|
||||
|
||||
private final TransactionTemplate transactionTemplate;
|
||||
|
||||
private ProducerListener<byte[], byte[]> producerListener;
|
||||
|
||||
private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties();
|
||||
|
||||
public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties,
|
||||
KafkaTopicProvisioner provisioningProvider) {
|
||||
super(false, headersToMap(configurationProperties), provisioningProvider);
|
||||
super(true, null, provisioningProvider);
|
||||
this.configurationProperties = configurationProperties;
|
||||
}
|
||||
|
||||
private static String[] headersToMap(KafkaBinderConfigurationProperties configurationProperties) {
|
||||
String[] headersToMap;
|
||||
if (ObjectUtils.isEmpty(configurationProperties.getHeaders())) {
|
||||
headersToMap = BinderHeaders.STANDARD_HEADERS;
|
||||
if (StringUtils.hasText(configurationProperties.getTransaction().getTransactionIdPrefix())) {
|
||||
this.transactionManager = new KafkaTransactionManager<>(
|
||||
getProducerFactory(configurationProperties.getTransaction().getTransactionIdPrefix(),
|
||||
new ExtendedProducerProperties<>(configurationProperties.getTransaction().getProducer())));
|
||||
this.transactionTemplate = new TransactionTemplate(this.transactionManager);
|
||||
}
|
||||
else {
|
||||
String[] combinedHeadersToMap = Arrays.copyOfRange(BinderHeaders.STANDARD_HEADERS, 0,
|
||||
BinderHeaders.STANDARD_HEADERS.length + configurationProperties.getHeaders().length);
|
||||
System.arraycopy(configurationProperties.getHeaders(), 0, combinedHeadersToMap,
|
||||
BinderHeaders.STANDARD_HEADERS.length,
|
||||
configurationProperties.getHeaders().length);
|
||||
headersToMap = combinedHeadersToMap;
|
||||
this.transactionManager = null;
|
||||
this.transactionTemplate = null;
|
||||
}
|
||||
return headersToMap;
|
||||
}
|
||||
|
||||
public void setExtendedBindingProperties(KafkaExtendedBindingProperties extendedBindingProperties) {
|
||||
@@ -165,7 +178,16 @@ public class KafkaMessageChannelBinder extends
|
||||
protected MessageHandler createProducerMessageHandler(final ProducerDestination destination,
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties, MessageChannel errorChannel)
|
||||
throws Exception {
|
||||
final DefaultKafkaProducerFactory<byte[], byte[]> producerFB = getProducerFactory(producerProperties);
|
||||
/*
|
||||
* IMPORTANT: With a transactional binder, individual producer properties for
|
||||
* Kafka are ignored; the global binder
|
||||
* (spring.cloud.stream.kafka.binder.transaction.producer.*) properties are used
|
||||
* instead, for all producers. A binder is transactional when
|
||||
* 'spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix' has text.
|
||||
*/
|
||||
final ProducerFactory<byte[], byte[]> producerFB = this.transactionManager != null
|
||||
? this.transactionManager.getProducerFactory()
|
||||
: getProducerFactory(null, producerProperties);
|
||||
Collection<PartitionInfo> partitions = provisioningProvider.getPartitionsForTopic(
|
||||
producerProperties.getPartitionCount(),
|
||||
false,
|
||||
@@ -176,7 +198,7 @@ public class KafkaMessageChannelBinder extends
|
||||
Producer<byte[], byte[]> producer = producerFB.createProducer();
|
||||
List<PartitionInfo> partitionsFor = producer.partitionsFor(destination.getName());
|
||||
producer.close();
|
||||
producerFB.destroy();
|
||||
((DisposableBean) producerFB).destroy();
|
||||
return partitionsFor;
|
||||
}
|
||||
|
||||
@@ -208,10 +230,21 @@ public class KafkaMessageChannelBinder extends
|
||||
if (errorChannel != null) {
|
||||
handler.setSendFailureChannel(errorChannel);
|
||||
}
|
||||
String[] headerPatterns = producerProperties.getExtension().getHeaderPatterns();
|
||||
if (headerPatterns != null && headerPatterns.length > 0) {
|
||||
List<String> patterns = new LinkedList<>(Arrays.asList(headerPatterns));
|
||||
if (!patterns.contains("!" + MessageHeaders.TIMESTAMP)) {
|
||||
patterns.add(0, "!" + MessageHeaders.TIMESTAMP);
|
||||
}
|
||||
if (!patterns.contains("!" + MessageHeaders.ID)) {
|
||||
patterns.add(0, "!" + MessageHeaders.ID);
|
||||
}
|
||||
handler.setHeaderMapper(new DefaultKafkaHeaderMapper(patterns.toArray(new String[patterns.size()])));
|
||||
}
|
||||
return handler;
|
||||
}
|
||||
|
||||
private DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(
|
||||
protected DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(String transactionIdPrefix,
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
|
||||
Map<String, Object> props = new HashMap<>();
|
||||
props.put(ProducerConfig.RETRIES_CONFIG, 0);
|
||||
@@ -240,7 +273,11 @@ public class KafkaMessageChannelBinder extends
|
||||
if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) {
|
||||
props.putAll(producerProperties.getExtension().getConfiguration());
|
||||
}
|
||||
return new DefaultKafkaProducerFactory<>(props);
|
||||
DefaultKafkaProducerFactory<byte[], byte[]> producerFactory = new DefaultKafkaProducerFactory<>(props);
|
||||
if (transactionIdPrefix != null) {
|
||||
producerFactory.setTransactionIdPrefix(transactionIdPrefix);
|
||||
}
|
||||
return producerFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -297,6 +334,9 @@ public class KafkaMessageChannelBinder extends
|
||||
|| extendedConsumerProperties.getExtension().isAutoRebalanceEnabled()
|
||||
? new ContainerProperties(destination.getName())
|
||||
: new ContainerProperties(topicPartitionInitialOffsets);
|
||||
if (this.transactionManager != null) {
|
||||
containerProperties.setTransactionManager(this.transactionManager);
|
||||
}
|
||||
int concurrency = Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size());
|
||||
@SuppressWarnings("rawtypes")
|
||||
final ConcurrentMessageListenerContainer<?, ?> messageListenerContainer =
|
||||
@@ -325,6 +365,50 @@ public class KafkaMessageChannelBinder extends
|
||||
}
|
||||
final KafkaMessageDrivenChannelAdapter<?, ?> kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>(
|
||||
messageListenerContainer);
|
||||
MessagingMessageConverter messageConverter;
|
||||
if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) {
|
||||
messageConverter = new MessagingMessageConverter();
|
||||
StandardHeaders standardHeaders = extendedConsumerProperties.getExtension().getStandardHeaders();
|
||||
messageConverter.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders)
|
||||
|| StandardHeaders.both.equals(standardHeaders));
|
||||
messageConverter.setGenerateTimestamp(StandardHeaders.timestamp.equals(standardHeaders)
|
||||
|| StandardHeaders.both.equals(standardHeaders));
|
||||
}
|
||||
else {
|
||||
try {
|
||||
messageConverter = getApplicationContext().getBean(
|
||||
extendedConsumerProperties.getExtension().getConverterBeanName(),
|
||||
MessagingMessageConverter.class);
|
||||
}
|
||||
catch (NoSuchBeanDefinitionException e) {
|
||||
throw new IllegalStateException("Converter bean not present in application context", e);
|
||||
}
|
||||
}
|
||||
KafkaHeaderMapper mapper = null;
|
||||
if (this.configurationProperties.getHeaderMapperBeanName() != null) {
|
||||
mapper = getApplicationContext().getBean(this.configurationProperties.getHeaderMapperBeanName(),
|
||||
KafkaHeaderMapper.class);
|
||||
}
|
||||
if (mapper == null) {
|
||||
DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper() {
|
||||
|
||||
@Override
|
||||
public void toHeaders(Headers source, Map<String, Object> headers) {
|
||||
super.toHeaders(source, headers);
|
||||
if (headers.size() > 0) {
|
||||
headers.put("scst_nativeHeadersPresent", Boolean.TRUE);
|
||||
}
|
||||
}
|
||||
|
||||
};
|
||||
String[] trustedPackages = extendedConsumerProperties.getExtension().getTrustedPackages();
|
||||
if (!StringUtils.isEmpty(trustedPackages)) {
|
||||
headerMapper.addTrustedPackages(trustedPackages);
|
||||
}
|
||||
mapper = headerMapper;
|
||||
}
|
||||
messageConverter.setHeaderMapper(mapper);
|
||||
kafkaMessageDrivenChannelAdapter.setMessageConverter(messageConverter);
|
||||
kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());
|
||||
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, consumerGroup,
|
||||
extendedConsumerProperties);
|
||||
@@ -338,6 +422,7 @@ public class KafkaMessageChannelBinder extends
|
||||
return kafkaMessageDrivenChannelAdapter;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
protected ErrorMessageStrategy getErrorMessageStrategy() {
|
||||
return new RawRecordHeaderErrorMessageStrategy();
|
||||
@@ -347,8 +432,9 @@ public class KafkaMessageChannelBinder extends
|
||||
protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination, final String group,
|
||||
final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
|
||||
if (extendedConsumerProperties.getExtension().isEnableDlq()) {
|
||||
DefaultKafkaProducerFactory<byte[], byte[]> producerFactory = getProducerFactory(
|
||||
new ExtendedProducerProperties<>(new KafkaProducerProperties()));
|
||||
ProducerFactory<byte[], byte[]> producerFactory = this.transactionManager != null
|
||||
? this.transactionManager.getProducerFactory()
|
||||
: getProducerFactory(null, new ExtendedProducerProperties<>(new KafkaProducerProperties()));
|
||||
final KafkaTemplate<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFactory);
|
||||
return new MessageHandler() {
|
||||
|
||||
@@ -359,38 +445,25 @@ public class KafkaMessageChannelBinder extends
|
||||
final byte[] key = record.key() != null
|
||||
? Utils.toArray(ByteBuffer.wrap((byte[]) record.key()))
|
||||
: null;
|
||||
final byte[] payload;
|
||||
if (HeaderMode.embeddedHeaders == extendedConsumerProperties.getHeaderMode()
|
||||
&& message.getPayload() instanceof Throwable) {
|
||||
final byte[] payload = record.value() != null
|
||||
? Utils.toArray(ByteBuffer.wrap((byte[]) record.value())) : null;
|
||||
if (message.getPayload() instanceof Throwable) {
|
||||
final Throwable throwable = (Throwable) message.getPayload();
|
||||
final String failureMessage = throwable.getMessage();
|
||||
|
||||
try {
|
||||
MessageValues messageValues = EmbeddedHeaderUtils
|
||||
.extractHeaders(MessageBuilder.withPayload((byte[]) record.value()).build(),
|
||||
false);
|
||||
messageValues.put(X_ORIGINAL_TOPIC, record.topic());
|
||||
messageValues.put(X_EXCEPTION_MESSAGE, failureMessage);
|
||||
messageValues.put(X_EXCEPTION_STACKTRACE, getStackTraceAsString(throwable));
|
||||
|
||||
final String[] headersToEmbed = new ArrayList<>(messageValues.keySet()).toArray(
|
||||
new String[messageValues.keySet().size()]);
|
||||
payload = EmbeddedHeaderUtils.embedHeaders(messageValues,
|
||||
EmbeddedHeaderUtils.headersToEmbed(headersToEmbed));
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
else {
|
||||
payload = record.value() != null
|
||||
? Utils.toArray(ByteBuffer.wrap((byte[]) record.value())) : null;
|
||||
record.headers()
|
||||
.add(new RecordHeader(X_ORIGINAL_TOPIC, record.topic().getBytes(StandardCharsets.UTF_8)));
|
||||
record.headers()
|
||||
.add(new RecordHeader(X_EXCEPTION_MESSAGE, failureMessage.getBytes(StandardCharsets.UTF_8)));
|
||||
record.headers()
|
||||
.add(new RecordHeader(X_EXCEPTION_STACKTRACE, getStackTraceAsString(throwable).getBytes()));
|
||||
}
|
||||
String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName())
|
||||
? extendedConsumerProperties.getExtension().getDlqName()
|
||||
: "error." + destination.getName() + "." + group;
|
||||
ListenableFuture<SendResult<byte[], byte[]>> sentDlq = kafkaTemplate.send(dlqName,
|
||||
record.partition(), key, payload);
|
||||
ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>(dlqName, record.partition(),
|
||||
key, payload, record.headers());
|
||||
ListenableFuture<SendResult<byte[], byte[]>> sentDlq = kafkaTemplate.send(producerRecord);
|
||||
sentDlq.addCallback(new ListenableFutureCallback<SendResult<byte[], byte[]>>() {
|
||||
StringBuilder sb = new StringBuilder().append(" a message with key='")
|
||||
.append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'")
|
||||
@@ -485,11 +558,11 @@ public class KafkaMessageChannelBinder extends
|
||||
|
||||
private boolean running = true;
|
||||
|
||||
private final DefaultKafkaProducerFactory<byte[], byte[]> producerFactory;
|
||||
private final ProducerFactory<byte[], byte[]> producerFactory;
|
||||
|
||||
ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic,
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
|
||||
DefaultKafkaProducerFactory<byte[], byte[]> producerFactory) {
|
||||
ProducerFactory<byte[], byte[]> producerFactory) {
|
||||
super(kafkaTemplate);
|
||||
setTopicExpression(new LiteralExpression(topic));
|
||||
setMessageKeyExpression(producerProperties.getExtension().getMessageKeyExpression());
|
||||
@@ -517,7 +590,9 @@ public class KafkaMessageChannelBinder extends
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
producerFactory.stop();
|
||||
if (this.producerFactory instanceof Lifecycle) {
|
||||
((Lifecycle) producerFactory).stop();
|
||||
}
|
||||
this.running = false;
|
||||
}
|
||||
|
||||
@@ -525,6 +600,30 @@ public class KafkaMessageChannelBinder extends
|
||||
public boolean isRunning() {
|
||||
return this.running;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void handleMessageInternal(final Message<?> message) throws Exception {
|
||||
if (KafkaMessageChannelBinder.this.transactionTemplate != null
|
||||
&& !TransactionSynchronizationManager.isActualTransactionActive()) {
|
||||
KafkaMessageChannelBinder.this.transactionTemplate.execute(
|
||||
new TransactionCallback<Object>() {
|
||||
@Override
|
||||
public Object doInTransaction(TransactionStatus transactionStatus) {
|
||||
try {
|
||||
ProducerConfigurationMessageHandler.super.handleMessageInternal(message);
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new KafkaException("Exception on transactional send", e);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
});
|
||||
}
|
||||
else {
|
||||
super.handleMessageInternal(message);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class TopicInformation {
|
||||
@@ -34,7 +34,6 @@ import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoCon
|
||||
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.KafkaBinderJaasInitializerListener;
|
||||
import org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics;
|
||||
import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
|
||||
import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation;
|
||||
@@ -46,7 +45,6 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBin
|
||||
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
|
||||
import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.ApplicationListener;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Condition;
|
||||
import org.springframework.context.annotation.ConditionContext;
|
||||
@@ -57,6 +55,7 @@ import org.springframework.core.type.AnnotatedTypeMetadata;
|
||||
import org.springframework.integration.codec.Codec;
|
||||
import org.springframework.kafka.core.ConsumerFactory;
|
||||
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
|
||||
import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
|
||||
import org.springframework.kafka.support.LoggingProducerListener;
|
||||
import org.springframework.kafka.support.ProducerListener;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
@@ -157,7 +156,7 @@ public class KafkaBinderConfiguration {
|
||||
}
|
||||
|
||||
@Bean(name = "adminUtilsOperation")
|
||||
@Conditional(Kafka10Present.class)
|
||||
@Conditional(Kafka1xPresent.class)
|
||||
@ConditionalOnClass(name = "kafka.admin.AdminUtils")
|
||||
public AdminUtilsOperation kafka10AdminUtilsOperation() {
|
||||
logger.info("AdminUtils selected: Kafka 0.10 AdminUtils");
|
||||
@@ -165,15 +164,15 @@ public class KafkaBinderConfiguration {
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ApplicationListener<?> jaasInitializer() throws IOException {
|
||||
return new KafkaBinderJaasInitializerListener();
|
||||
public KafkaJaasLoginModuleInitializer jaasInitializer() throws IOException {
|
||||
return new KafkaJaasLoginModuleInitializer();
|
||||
}
|
||||
|
||||
static class Kafka10Present implements Condition {
|
||||
static class Kafka1xPresent implements Condition {
|
||||
|
||||
@Override
|
||||
public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) {
|
||||
return AppInfoParser.getVersion().startsWith("0.10");
|
||||
return AppInfoParser.getVersion().startsWith("0.1");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -66,10 +66,10 @@ public class KafkaBinderAutoConfigurationPropertiesTest {
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(
|
||||
new KafkaProducerProperties());
|
||||
Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory",
|
||||
ExtendedProducerProperties.class);
|
||||
String.class, ExtendedProducerProperties.class);
|
||||
getProducerFactoryMethod.setAccessible(true);
|
||||
DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod
|
||||
.invoke(this.kafkaMessageChannelBinder, producerProperties);
|
||||
.invoke(this.kafkaMessageChannelBinder, "foo", producerProperties);
|
||||
Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs",
|
||||
Map.class);
|
||||
ReflectionUtils.makeAccessible(producerFactoryConfigField);
|
||||
@@ -63,10 +63,10 @@ public class KafkaBinderConfigurationPropertiesTest {
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(
|
||||
kafkaProducerProperties);
|
||||
Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory",
|
||||
ExtendedProducerProperties.class);
|
||||
String.class, ExtendedProducerProperties.class);
|
||||
getProducerFactoryMethod.setAccessible(true);
|
||||
DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod
|
||||
.invoke(this.kafkaMessageChannelBinder, producerProperties);
|
||||
.invoke(this.kafkaMessageChannelBinder, "bar", producerProperties);
|
||||
Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs",
|
||||
Map.class);
|
||||
ReflectionUtils.makeAccessible(producerFactoryConfigField);
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2017 the original author or authors.
|
||||
* Copyright 2016-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -16,12 +16,6 @@
|
||||
|
||||
package org.springframework.cloud.stream.binder.kafka;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.fail;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Arrays;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
@@ -33,9 +27,15 @@ import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
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.KafkaConsumer;
|
||||
import org.apache.kafka.clients.producer.ProducerConfig;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
|
||||
import org.apache.kafka.common.serialization.ByteArraySerializer;
|
||||
import org.apache.kafka.common.serialization.LongDeserializer;
|
||||
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
@@ -55,11 +55,13 @@ import org.springframework.cloud.stream.binder.PartitionTestSupport;
|
||||
import org.springframework.cloud.stream.binder.TestUtils;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders;
|
||||
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.kafka.utils.KafkaTopicUtils;
|
||||
import org.springframework.cloud.stream.config.BindingProperties;
|
||||
import org.springframework.cloud.stream.provisioning.ProvisioningException;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
@@ -76,6 +78,7 @@ import org.springframework.kafka.support.Acknowledgment;
|
||||
import org.springframework.kafka.support.KafkaHeaders;
|
||||
import org.springframework.kafka.support.SendResult;
|
||||
import org.springframework.kafka.support.TopicPartitionInitialOffset;
|
||||
import org.springframework.kafka.support.converter.MessagingMessageConverter;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
@@ -91,8 +94,10 @@ import org.springframework.retry.support.RetryTemplate;
|
||||
import org.springframework.util.concurrent.ListenableFuture;
|
||||
import org.springframework.util.concurrent.SettableListenableFuture;
|
||||
|
||||
import kafka.utils.ZKStringSerializer$;
|
||||
import kafka.utils.ZkUtils;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.fail;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
/**
|
||||
* @author Soby Chacko
|
||||
@@ -143,20 +148,15 @@ public abstract class KafkaBinderTests extends
|
||||
|
||||
@Test
|
||||
public void testDlqAndRetry() throws Exception {
|
||||
testDlqGuts(true, false);
|
||||
testDlqGuts(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDlq() throws Exception {
|
||||
testDlqGuts(false, false);
|
||||
testDlqGuts(false);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDlqRawHeaders() throws Exception {
|
||||
testDlqGuts(false, true);
|
||||
}
|
||||
|
||||
private void testDlqGuts(boolean withRetry, boolean raw) throws Exception {
|
||||
private void testDlqGuts(boolean withRetry) throws Exception {
|
||||
AbstractKafkaTestBinder binder = getBinder();
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
DirectChannel moduleInputChannel = new DirectChannel();
|
||||
@@ -165,18 +165,14 @@ public abstract class KafkaBinderTests extends
|
||||
moduleInputChannel.subscribe(handler);
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
|
||||
producerProperties.setPartitionCount(2);
|
||||
if (raw) {
|
||||
producerProperties.setHeaderMode(HeaderMode.raw);
|
||||
}
|
||||
producerProperties.getExtension().setHeaderPatterns(new String[] { MessageHeaders.CONTENT_TYPE,
|
||||
"dlqTestHeader" });
|
||||
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
|
||||
consumerProperties.setMaxAttempts(withRetry ? 2 : 1);
|
||||
consumerProperties.setBackOffInitialInterval(100);
|
||||
consumerProperties.setBackOffMaxInterval(150);
|
||||
consumerProperties.getExtension().setEnableDlq(true);
|
||||
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
|
||||
if (raw) {
|
||||
consumerProperties.setHeaderMode(HeaderMode.raw);
|
||||
}
|
||||
long uniqueBindingId = System.currentTimeMillis();
|
||||
|
||||
String producerName = "dlqTest." + uniqueBindingId + ".0";
|
||||
@@ -187,9 +183,6 @@ public abstract class KafkaBinderTests extends
|
||||
|
||||
ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
|
||||
dlqConsumerProperties.setMaxAttempts(1);
|
||||
if (raw) {
|
||||
dlqConsumerProperties.setHeaderMode(HeaderMode.raw);
|
||||
}
|
||||
|
||||
ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(), "applicationContext",
|
||||
ApplicationContext.class);
|
||||
@@ -229,21 +222,17 @@ public abstract class KafkaBinderTests extends
|
||||
|
||||
Message<?> receivedMessage = receive(dlqChannel, 3);
|
||||
assertThat(receivedMessage).isNotNull();
|
||||
if (raw) {
|
||||
assertThat(new String((byte[]) receivedMessage.getPayload(), StandardCharsets.UTF_8))
|
||||
.isEqualTo(testMessagePayload);
|
||||
}
|
||||
else {
|
||||
assertThat(receivedMessage.getPayload()).isEqualTo(testMessagePayload);
|
||||
final MessageHeaders headers = receivedMessage.getHeaders();
|
||||
assertThat(headers.get(KafkaMessageChannelBinder.X_ORIGINAL_TOPIC)).isEqualTo(producerName);
|
||||
assertThat(headers.get(KafkaMessageChannelBinder.X_EXCEPTION_MESSAGE))
|
||||
.isEqualTo("failed to send Message to channel 'null'; nested exception is java.lang.RuntimeException: fail");
|
||||
assertThat(headers.get(KafkaMessageChannelBinder.X_EXCEPTION_STACKTRACE)).isNotNull();
|
||||
assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isNotNull();
|
||||
assertThat(headers.get("dlqTestHeader")).isEqualTo("propagatedToDlq");
|
||||
assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
|
||||
}
|
||||
assertThat(receivedMessage.getPayload()).isEqualTo(testMessagePayload);
|
||||
final MessageHeaders headers = receivedMessage.getHeaders();
|
||||
assertThat(headers.get(KafkaMessageChannelBinder.X_ORIGINAL_TOPIC)).isEqualTo(producerName.getBytes());
|
||||
assertThat(headers.get(KafkaMessageChannelBinder.X_EXCEPTION_MESSAGE))
|
||||
.isEqualTo(
|
||||
"failed to send Message to channel 'null'; nested exception is java.lang.RuntimeException: fail"
|
||||
.getBytes());
|
||||
assertThat(headers.get(KafkaMessageChannelBinder.X_EXCEPTION_STACKTRACE)).isNotNull();
|
||||
assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isNotNull();
|
||||
assertThat(headers.get("dlqTestHeader")).isEqualTo("propagatedToDlq");
|
||||
assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
|
||||
binderBindUnbindLatency();
|
||||
|
||||
// verify we got a message on the dedicated error channel and the global (via bridge)
|
||||
@@ -1438,7 +1427,7 @@ public abstract class KafkaBinderTests extends
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
public void testNativeSerializationWithCustomSerializerDeserializer() throws Exception {
|
||||
Binding<?> producerBinding = null;
|
||||
Binding<?> consumerBinding = null;
|
||||
@@ -1466,6 +1455,7 @@ public abstract class KafkaBinderTests extends
|
||||
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
|
||||
consumerProperties.getExtension().getConfiguration().put("value.deserializer",
|
||||
"org.apache.kafka.common.serialization.IntegerDeserializer");
|
||||
consumerProperties.getExtension().setStandardHeaders(StandardHeaders.both);
|
||||
consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties);
|
||||
// Let the consumer actually bind to the producer before sending a msg
|
||||
binderBindUnbindLatency();
|
||||
@@ -1474,6 +1464,70 @@ public abstract class KafkaBinderTests extends
|
||||
assertThat(inbound).isNotNull();
|
||||
assertThat(inbound.getPayload()).isEqualTo(10);
|
||||
assertThat(inbound.getHeaders()).doesNotContainKey("contentType");
|
||||
assertThat(inbound.getHeaders().getId()).isNotNull();
|
||||
assertThat(inbound.getHeaders().getTimestamp()).isNotNull();
|
||||
}
|
||||
finally {
|
||||
if (producerBinding != null) {
|
||||
producerBinding.unbind();
|
||||
}
|
||||
if (consumerBinding != null) {
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
public void testNativeSerializationWithCustomSerializerDeserializerBytesPayload() throws Exception {
|
||||
Binding<?> producerBinding = null;
|
||||
Binding<?> consumerBinding = null;
|
||||
try {
|
||||
byte[] testPayload = new byte[1];
|
||||
Message<?> message = MessageBuilder.withPayload(testPayload)
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, "something/funky")
|
||||
.build();
|
||||
SubscribableChannel moduleOutputChannel = new DirectChannel();
|
||||
String testTopicName = "existing" + System.currentTimeMillis();
|
||||
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
|
||||
final ZkClient zkClient;
|
||||
zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
|
||||
configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(),
|
||||
ZKStringSerializer$.MODULE$);
|
||||
final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
|
||||
invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties());
|
||||
configurationProperties.setAutoAddPartitions(true);
|
||||
Binder binder = getBinder(configurationProperties);
|
||||
ConfigurableApplicationContext context = TestUtils.getPropertyValue(binder, "binder.applicationContext",
|
||||
ConfigurableApplicationContext.class);
|
||||
MessagingMessageConverter converter = new MessagingMessageConverter();
|
||||
converter.setGenerateMessageId(true);
|
||||
converter.setGenerateTimestamp(true);
|
||||
context.getBeanFactory().registerSingleton("testConverter", converter);
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
|
||||
producerProperties.setUseNativeEncoding(true);
|
||||
producerProperties.getExtension()
|
||||
.getConfiguration()
|
||||
.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
|
||||
producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties);
|
||||
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
|
||||
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
|
||||
consumerProperties.getExtension()
|
||||
.getConfiguration()
|
||||
.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
|
||||
consumerProperties.getExtension().setConverterBeanName("testConverter");
|
||||
consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties);
|
||||
// Let the consumer actually bind to the producer before sending a msg
|
||||
binderBindUnbindLatency();
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = receive(moduleInputChannel, 500);
|
||||
assertThat(inbound).isNotNull();
|
||||
assertThat(inbound.getPayload()).isEqualTo(new byte[1]);
|
||||
assertThat(inbound.getHeaders()).containsKey("contentType");
|
||||
assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("something/funky");
|
||||
assertThat(inbound.getHeaders().getId()).isNotNull();
|
||||
assertThat(inbound.getHeaders().getTimestamp()).isNotNull();
|
||||
}
|
||||
finally {
|
||||
if (producerBinding != null) {
|
||||
@@ -1643,7 +1697,7 @@ public abstract class KafkaBinderTests extends
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
public void testSendAndReceiveWithRawMode() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
@@ -1816,7 +1870,7 @@ public abstract class KafkaBinderTests extends
|
||||
new DirectFieldAccessor(endpoint).setPropertyValue("kafkaTemplate",
|
||||
new KafkaTemplate(mock(ProducerFactory.class)) {
|
||||
|
||||
@Override
|
||||
@Override // SIK < 2.3
|
||||
public ListenableFuture<SendResult> send(String topic, Object payload) {
|
||||
sent.set(payload);
|
||||
SettableListenableFuture<SendResult> future = new SettableListenableFuture<>();
|
||||
@@ -1824,6 +1878,14 @@ public abstract class KafkaBinderTests extends
|
||||
return future;
|
||||
}
|
||||
|
||||
@Override // SIK 2.3+
|
||||
public ListenableFuture<SendResult> send(ProducerRecord record) {
|
||||
sent.set(record.value());
|
||||
SettableListenableFuture<SendResult> future = new SettableListenableFuture<>();
|
||||
future.setException(fooException);
|
||||
return future;
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
moduleOutputChannel.send(message);
|
||||
@@ -4,6 +4,7 @@
|
||||
<pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
<logger name="org.apache.kafka" level="DEBUG"/>
|
||||
<logger name="org.springframework.integration.kafka" level="INFO"/>
|
||||
<logger name="org.springframework.kafka" level="INFO"/>
|
||||
<logger name="org.springframework.cloud.stream" level="INFO" />
|
||||
@@ -2,21 +2,21 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>spring-cloud-stream-binder-kstream</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kstream11</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>spring-cloud-stream-binder-kstream</name>
|
||||
<description>Kafka Streams Binder Implementation</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
|
||||
<version>1.3.4.RELEASE</version>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-parent</artifactId>
|
||||
<version>1.3.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka11-core</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
@@ -54,7 +54,7 @@ public class KStreamBinder extends
|
||||
AbstractBinder<KStream<Object, Object>, ExtendedConsumerProperties<KStreamConsumerProperties>, ExtendedProducerProperties<KStreamProducerProperties>>
|
||||
implements ExtendedPropertiesBinder<KStream<Object, Object>, KStreamConsumerProperties, KStreamProducerProperties> {
|
||||
|
||||
private String[] headers;
|
||||
private final String[] headers;
|
||||
|
||||
private final KafkaTopicProvisioner kafkaTopicProvisioner;
|
||||
|
||||
@@ -96,7 +96,7 @@ public class KStreamBinder extends
|
||||
public KeyValue<Object, Object> apply(Object k, Object v) {
|
||||
if (v instanceof Message) {
|
||||
try {
|
||||
return new KeyValue<>(k, (Object) KStreamBinder.this.serializeAndEmbedHeadersIfApplicable((Message<?>) v));
|
||||
return new KeyValue<>(k, KStreamBinder.this.serializeAndEmbedHeadersIfApplicable((Message<?>) v));
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new IllegalArgumentException(e);
|
||||
@@ -123,7 +123,7 @@ public class KStreamBinder extends
|
||||
.map(new KeyValueMapper<Object, Object, KeyValue<Object, Object>>() {
|
||||
@Override
|
||||
public KeyValue<Object, Object> apply(Object k, Object v) {
|
||||
return KeyValue.pair(k, ((Message<?>) v).getPayload());
|
||||
return KeyValue.pair(k, ((Message<Object>) v).getPayload());
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -167,7 +167,7 @@ public class KStreamBinder extends
|
||||
return new DefaultBinding<>(name, null, outboundBindTarget, null);
|
||||
}
|
||||
|
||||
private byte[] serializeAndEmbedHeadersIfApplicable(Message<?> message) throws Exception {
|
||||
private Object serializeAndEmbedHeadersIfApplicable(Message<?> message) throws Exception {
|
||||
MessageValues transformed = serializePayloadIfNecessary(message);
|
||||
byte[] payload;
|
||||
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user