diff --git a/pom.xml b/pom.xml index 9fd6d8b0d..b8bb94316 100644 --- a/pom.xml +++ b/pom.xml @@ -21,7 +21,6 @@ spring-cloud-stream-binder-kafka spring-cloud-starter-stream-kafka spring-cloud-stream-binder-kafka-docs - spring-cloud-stream-binder-kafka-test spring-cloud-stream-binder-kafka-core spring-cloud-stream-binder-kstream @@ -106,13 +105,6 @@ - - org.springframework.cloud - spring-cloud-stream-binder-kafka - ${project.version} - test-jar - test - diff --git a/spring-cloud-stream-binder-kafka-test/.jdk8 b/spring-cloud-stream-binder-kafka-test/.jdk8 deleted file mode 100644 index e69de29bb..000000000 diff --git a/spring-cloud-stream-binder-kafka-test/pom.xml b/spring-cloud-stream-binder-kafka-test/pom.xml deleted file mode 100644 index b80169b03..000000000 --- a/spring-cloud-stream-binder-kafka-test/pom.xml +++ /dev/null @@ -1,135 +0,0 @@ - - - 4.0.0 - - org.springframework.cloud - spring-cloud-stream-binder-kafka-parent - 2.0.0.BUILD-SNAPSHOT - - spring-cloud-stream-binder-kafka-test - Spring Cloud Stream Kafka Binder Tests - http://projects.spring.io/spring-cloud - - Pivotal Software, Inc. - http://www.spring.io - - - ${basedir}/../.. - 0.11.0.0 - 2.0.1.BUILD-SNAPSHOT - - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - test - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - test-jar - test - - - org.springframework.kafka - spring-kafka - test - - - org.apache.kafka - kafka_2.11 - test - - - org.apache.kafka - kafka-clients - test - - - org.springframework.kafka - spring-kafka-test - test - - - org.springframework.integration - spring-integration-kafka - - - org.springframework.integration - spring-integration-core - - - org.springframework.integration - spring-integration-jmx - - - org.springframework.cloud - spring-cloud-stream-binder-test - test - - - org.springframework.cloud - spring-cloud-stream-schema - ${spring-cloud-stream.version} - test - - - io.confluent - kafka-avro-serializer - 3.3.0 - test - - - io.confluent - kafka-schema-registry - 3.3.0 - test - - - org.glassfish.jersey.inject - jersey-hk2 - 2.26-b06 - - - org.glassfish.jersey.bundles.repackaged - jersey-guava - 2.6 - - - - - - spring-snapshots - Spring Snapshots - http://repo.spring.io/libs-snapshot-local - - true - - - false - - - - spring-milestones - Spring Milestones - http://repo.spring.io/libs-milestone-local - - false - - - - spring-releases - Spring Releases - http://repo.spring.io/release - - false - - - - confluent - http://packages.confluent.io/maven/ - - - - diff --git a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java deleted file mode 100644 index 07063b150..000000000 --- a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ /dev/null @@ -1,366 +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 - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.UUID; - -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.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; -import org.springframework.cloud.stream.binder.Spy; -import org.springframework.cloud.stream.binder.kafka.admin.KafkaAdminUtilsOperation; -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; -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.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.assertj.core.api.Assertions.assertThat; -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 KafkaBinderTests extends AbstractKafkaBinderTests { - - private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); - - @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); - - private KafkaTestBinder binder; - - private final KafkaAdminUtilsOperation adminUtilsOperation = new KafkaAdminUtilsOperation(); - - @Override - protected void binderBindUnbindLatency() throws InterruptedException { - Thread.sleep(500); - } - - @Override - protected KafkaTestBinder getBinder() { - if (binder == null) { - KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binder = new KafkaTestBinder(binderConfiguration); - } - return binder; - } - - @Override - protected KafkaBinderConfigurationProperties createConfigurationProperties() { - KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); - BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); - List 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 KafkaTestBinder(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 consumerFactory() { - Map 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 valueDecoder = new ByteArrayDeserializer(); - Deserializer keyDecoder = new ByteArrayDeserializer(); - - return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); - } - - @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 producerBinding = binder.bindProducer("bar.0", moduleOutputChannel, - producerBindingProperties.getProducer()); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setTrustedPackages(new String[]{"org.springframework.util"}); - Binding 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_VALUE) - .setHeader("foo", MimeTypeUtils.TEXT_PLAIN) - .build(); - - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - Assertions.assertThat(inbound).isNotNull(); - Assertions.assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); - Assertions.assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull(); - Assertions.assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)) - .isEqualTo(MimeTypeUtils.TEXT_PLAIN); - 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 { - 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 schemaRegistryProps = new HashMap<>(); - schemaRegistryProps.put("kafkastore.connection.url", configurationProperties.getZkConnectionString()); - schemaRegistryProps.put("listeners", "http://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 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 producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties); - ExtendedConsumerProperties 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 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(); - } - - @Override - public void testSendAndReceiveWithExplicitConsumerGroupWithRawMode() { - // raw mode no longer needed - } - - @Override - public void testSendAndReceiveWithRawModeAndStringPayload() { - // raw mode no longer needed - } - - @Test - @Override - @SuppressWarnings("unchecked") - public void testSendAndReceiveNoOriginalContentType() throws Exception { - Binder binder = getBinder(); - - BindingProperties producerBindingProperties = createProducerBindingProperties( - createProducerProperties()); - DirectChannel moduleOutputChannel = createBindableChannel("output", - producerBindingProperties); - QueueChannel moduleInputChannel = new QueueChannel(); - Binding producerBinding = binder.bindProducer("bar.0", - moduleOutputChannel, producerBindingProperties.getProducer()); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setTrustedPackages(new String[] {"org.springframework.util"}); - Binding consumerBinding = binder.bindConsumer("bar.0", - "testSendAndReceiveNoOriginalContentType", moduleInputChannel, - consumerProperties); - binderBindUnbindLatency(); - - //TODO: Will have to fix the MimeType to convert to byte array once this issue has been resolved: - //https://github.com/spring-projects/spring-kafka/issues/424 - Message message = org.springframework.integration.support.MessageBuilder.withPayload("foo".getBytes()) - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN_VALUE.getBytes()).build(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); - assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)) - .isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE.getBytes()); - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @Override - @SuppressWarnings("unchecked") - public void testSendAndReceive() throws Exception { - Binder binder = getBinder(); - BindingProperties outputBindingProperties = createProducerBindingProperties( - createProducerProperties()); - DirectChannel moduleOutputChannel = createBindableChannel("output", - outputBindingProperties); - QueueChannel moduleInputChannel = new QueueChannel(); - Binding producerBinding = binder.bindProducer("foo.0", - moduleOutputChannel, outputBindingProperties.getProducer()); - Binding consumerBinding = binder.bindConsumer("foo.0", - "testSendAndReceive", moduleInputChannel, createConsumerProperties()); - // Bypass conversion we are only testing sendReceive - Message message = org.springframework.integration.support.MessageBuilder.withPayload("foo".getBytes()) - .setHeader(MessageHeaders.CONTENT_TYPE, - MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE.getBytes()) - .build(); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); - assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)) - .isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE.getBytes()); - producerBinding.unbind(); - consumerBinding.unbind(); - } - -} diff --git a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java b/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java deleted file mode 100644 index b5c45c9b3..000000000 --- a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java +++ /dev/null @@ -1,85 +0,0 @@ -/* - * Copyright 2016 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka; - -import java.io.IOException; - -import org.apache.avro.Schema; -import org.apache.avro.reflect.Nullable; -import org.apache.avro.specific.SpecificRecordBase; - -import org.springframework.core.io.ClassPathResource; - -/** - * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan - */ -public class User1 extends SpecificRecordBase { - - @Nullable - private String name; - - @Nullable - private String favoriteColor; - - public String getName() { - return this.name; - } - - public void setName(String name) { - this.name = name; - } - - public String getFavoriteColor() { - return this.favoriteColor; - } - - public void setFavoriteColor(String favoriteColor) { - this.favoriteColor = favoriteColor; - } - - @Override - public Schema getSchema() { - try { - return new Schema.Parser().parse(new ClassPathResource("schemas/users_v1.schema").getInputStream()); - } - catch (IOException e) { - throw new IllegalStateException(e); - } - } - - @Override - public Object get(int i) { - if (i == 0) { - return getName().toString(); - } - if (i == 1) { - return getFavoriteColor().toString(); - } - return null; - } - - @Override - public void put(int i, Object o) { - if (i == 0) { - setName((String) o); - } - if (i == 1) { - setFavoriteColor((String) o); - } - } -} diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml index d0d143a16..4fd4134e7 100644 --- a/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binder-kafka/pom.xml @@ -17,7 +17,6 @@ org.springframework.cloud spring-cloud-stream-binder-kafka-core - 2.0.0.BUILD-SNAPSHOT org.springframework.boot @@ -71,28 +70,6 @@ spring-cloud-stream-binder-test test - - org.springframework.cloud - spring-cloud-stream-schema - ${spring-cloud-stream.version} - test - - - - - org.apache.maven.plugins - maven-jar-plugin - 3.0.2 - - - - test-jar - - - - - - diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java similarity index 91% rename from spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaBinderTests.java rename to spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 5cde2dd0c..c2c7edbdc 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -17,7 +17,9 @@ package org.springframework.cloud.stream.binder.kafka; import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -39,23 +41,30 @@ 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.Deserializer; import org.apache.kafka.common.serialization.LongDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; +import org.assertj.core.api.Assertions; import org.assertj.core.api.Condition; +import org.junit.Before; +import org.junit.ClassRule; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; import org.springframework.beans.DirectFieldAccessor; 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; import org.springframework.cloud.stream.binder.HeaderMode; import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; import org.springframework.cloud.stream.binder.PartitionTestSupport; +import org.springframework.cloud.stream.binder.Spy; import org.springframework.cloud.stream.binder.TestUtils; +import org.springframework.cloud.stream.binder.kafka.admin.KafkaAdminUtilsOperation; 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; @@ -71,6 +80,7 @@ import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.kafka.support.KafkaSendFailureException; +import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; @@ -79,6 +89,8 @@ 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.test.core.BrokerAddress; +import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -91,6 +103,7 @@ import org.springframework.messaging.support.MessageBuilder; import org.springframework.retry.backoff.FixedBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.SettableListenableFuture; @@ -107,12 +120,21 @@ import static org.mockito.Mockito.mock; * @author Henryk Konsek * @author Gary Russell */ -public abstract class AbstractKafkaBinderTests extends +public class KafkaBinderTests extends PartitionCapableBinderTests, ExtendedProducerProperties> { @Rule public ExpectedException expectedProvisioningException = ExpectedException.none(); + private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + + private KafkaTestBinder binder; + + private final KafkaAdminUtilsOperation adminUtilsOperation = new KafkaAdminUtilsOperation(); + @Override protected ExtendedConsumerProperties createConsumerProperties() { final ExtendedConsumerProperties kafkaConsumerProperties = new ExtendedConsumerProperties<>( @@ -131,21 +153,196 @@ public abstract class AbstractKafkaBinderTests extends return producerProperties; } - public abstract String getKafkaOffsetHeaderKey(); + @Override + protected void binderBindUnbindLatency() throws InterruptedException { + Thread.sleep(500); + } - protected abstract Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties); + @Override + protected KafkaTestBinder getBinder() { + if (binder == null) { + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + binder = new KafkaTestBinder(binderConfiguration); + } + return binder; + } - protected abstract KafkaBinderConfigurationProperties createConfigurationProperties(); + private KafkaBinderConfigurationProperties createConfigurationProperties() { + KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); + BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); + List 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; + } - protected abstract int partitionSize(String topic); + private int partitionSize(String topic) { + return consumerFactory().createConsumer().partitionsFor(topic).size(); + } - protected abstract ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties); + private ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { + final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(), + kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); - protected abstract void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, - int replicationFactor, Properties topicConfig); + return new ZkUtils(zkClient, null, false); + } - protected abstract int invokePartitionSize(String topic, - ZkUtils zkUtils); + private void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) { + adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties()); + } + + private int invokePartitionSize(String topic, ZkUtils zkUtils) { + return adminUtilsOperation.partitionSize(topic, zkUtils); + } + + private String getKafkaOffsetHeaderKey() { + return KafkaHeaders.OFFSET; + } + + private Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { + return new KafkaTestBinder(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 consumerFactory() { + Map 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 valueDecoder = new ByteArrayDeserializer(); + Deserializer keyDecoder = new ByteArrayDeserializer(); + + return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); + } + + @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 producerBinding = binder.bindProducer("bar.0", moduleOutputChannel, + producerBindingProperties.getProducer()); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setTrustedPackages(new String[]{"org.springframework.util"}); + Binding 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_VALUE) + .setHeader("foo", MimeTypeUtils.TEXT_PLAIN) + .build(); + + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + Assertions.assertThat(inbound).isNotNull(); + Assertions.assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); + Assertions.assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull(); + Assertions.assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MimeTypeUtils.TEXT_PLAIN); + 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(); + } + + @Test + @Override + @SuppressWarnings("unchecked") + public void testSendAndReceiveNoOriginalContentType() throws Exception { + Binder binder = getBinder(); + + BindingProperties producerBindingProperties = createProducerBindingProperties( + createProducerProperties()); + DirectChannel moduleOutputChannel = createBindableChannel("output", + producerBindingProperties); + QueueChannel moduleInputChannel = new QueueChannel(); + Binding producerBinding = binder.bindProducer("bar.0", + moduleOutputChannel, producerBindingProperties.getProducer()); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setTrustedPackages(new String[] {"org.springframework.util"}); + Binding consumerBinding = binder.bindConsumer("bar.0", + "testSendAndReceiveNoOriginalContentType", moduleInputChannel, + consumerProperties); + binderBindUnbindLatency(); + + //TODO: Will have to fix the MimeType to convert to byte array once this issue has been resolved: + //https://github.com/spring-projects/spring-kafka/issues/424 + Message message = org.springframework.integration.support.MessageBuilder.withPayload("foo".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN_VALUE.getBytes()).build(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + assertThat(inbound).isNotNull(); + assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); + assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE.getBytes()); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @Override + @SuppressWarnings("unchecked") + public void testSendAndReceive() throws Exception { + Binder binder = getBinder(); + BindingProperties outputBindingProperties = createProducerBindingProperties( + createProducerProperties()); + DirectChannel moduleOutputChannel = createBindableChannel("output", + outputBindingProperties); + QueueChannel moduleInputChannel = new QueueChannel(); + Binding producerBinding = binder.bindProducer("foo.0", + moduleOutputChannel, outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer("foo.0", + "testSendAndReceive", moduleInputChannel, createConsumerProperties()); + // Bypass conversion we are only testing sendReceive + Message message = org.springframework.integration.support.MessageBuilder.withPayload("foo".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, + MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE.getBytes()) + .build(); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + assertThat(inbound).isNotNull(); + assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); + assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE.getBytes()); + producerBinding.unbind(); + consumerBinding.unbind(); + } @Test public void testDlqAndRetry() throws Exception { @@ -1416,7 +1613,7 @@ public abstract class AbstractKafkaBinderTests extends Binding producerBinding = null; Binding consumerBinding = null; try { - Integer testPayload = new Integer(10); + Integer testPayload = 10; Message message = MessageBuilder.withPayload(testPayload).build(); SubscribableChannel moduleOutputChannel = new DirectChannel(); String testTopicName = "existing" + System.currentTimeMillis(); @@ -1518,7 +1715,7 @@ public abstract class AbstractKafkaBinderTests extends Binding producerBinding = null; Binding consumerBinding = null; try { - String testPayload = new String("test"); + String testPayload = "test"; Message message = MessageBuilder.withPayload(testPayload.getBytes()) .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN_VALUE.getBytes()) .build(); @@ -1697,113 +1894,6 @@ public abstract class AbstractKafkaBinderTests extends consumerBinding.unbind(); } - @Test - @SuppressWarnings("unchecked") - public void testSendAndReceiveWithRawModeAndStringPayload() throws Exception { - Binder binder = getBinder(); - DirectChannel moduleOutputChannel = new DirectChannel(); - QueueChannel moduleInputChannel = new QueueChannel(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setHeaderMode(HeaderMode.raw); - Binding producerBinding = binder.bindProducer("raw.string.0", moduleOutputChannel, - producerProperties); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.setHeaderMode(HeaderMode.raw); - Binding consumerBinding = binder.bindConsumer("raw.string.0", "test", moduleInputChannel, - consumerProperties); - Message message = org.springframework.integration.support.MessageBuilder - .withPayload("testSendAndReceiveWithRawModeAndStringPayload").build(); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat(new String((byte[]) inbound.getPayload())) - .isEqualTo("testSendAndReceiveWithRawModeAndStringPayload"); - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testSendAndReceiveWithExplicitConsumerGroupWithRawMode() throws Exception { - Binder binder = getBinder(); - DirectChannel moduleOutputChannel = new DirectChannel(); - // Test pub/sub by emulating how StreamPlugin handles taps - QueueChannel module1InputChannel = new QueueChannel(); - QueueChannel module2InputChannel = new QueueChannel(); - QueueChannel module3InputChannel = new QueueChannel(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setHeaderMode(HeaderMode.raw); - Binding producerBinding = binder.bindProducer("baz.raw.0", moduleOutputChannel, - producerProperties); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.setHeaderMode(HeaderMode.raw); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - Binding input1Binding = binder.bindConsumer("baz.raw.0", "test", module1InputChannel, - consumerProperties); - // A new module is using the tap as an input channel - String fooTapName = "baz.raw.0"; - Binding input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel, - consumerProperties); - // Another new module is using tap as an input channel - String barTapName = "baz.raw.0"; - Binding input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, - consumerProperties); - - Message message = org.springframework.integration.support.MessageBuilder - .withPayload("testSendAndReceiveWithExplicitConsumerGroupWithRawMode".getBytes()).build(); - boolean success = false; - boolean retried = false; - while (!success) { - moduleOutputChannel.send(message); - Message inbound = receive(module1InputChannel); - assertThat(inbound).isNotNull(); - assertThat(new String((byte[]) inbound.getPayload())) - .isEqualTo("testSendAndReceiveWithExplicitConsumerGroupWithRawMode"); - - Message tapped1 = receive(module2InputChannel); - Message tapped2 = receive(module3InputChannel); - if (tapped1 == null || tapped2 == null) { - // listener may not have started - assertThat(retried).isFalse().withFailMessage("Failed to receive tap after retry"); - retried = true; - continue; - } - success = true; - assertThat(new String((byte[]) tapped1.getPayload())) - .isEqualTo("testSendAndReceiveWithExplicitConsumerGroupWithRawMode"); - assertThat(new String((byte[]) tapped2.getPayload())) - .isEqualTo("testSendAndReceiveWithExplicitConsumerGroupWithRawMode"); - } - // delete one tap stream is deleted - input3Binding.unbind(); - Message message2 = org.springframework.integration.support.MessageBuilder.withPayload("bar".getBytes()) - .build(); - moduleOutputChannel.send(message2); - - // other tap still receives messages - Message tapped = receive(module2InputChannel); - assertThat(tapped).isNotNull(); - - // removed tap does not - assertThat(receive(module3InputChannel)).isNull(); - - // re-subscribed tap does receive the message - input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, createConsumerProperties()); - assertThat(receive(module3InputChannel)).isNotNull(); - - // clean up - input1Binding.unbind(); - input2Binding.unbind(); - input3Binding.unbind(); - producerBinding.unbind(); - assertThat(extractEndpoint(input1Binding).isRunning()).isFalse(); - assertThat(extractEndpoint(input2Binding).isRunning()).isFalse(); - assertThat(extractEndpoint(input3Binding).isRunning()).isFalse(); - assertThat(extractEndpoint(producerBinding).isRunning()).isFalse(); - } - @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testProducerErrorChannel() throws Exception { @@ -1819,25 +1909,13 @@ public abstract class AbstractKafkaBinderTests extends SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors", SubscribableChannel.class); final AtomicReference> errorMessage = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(2); - ec.subscribe(new MessageHandler() { - - @Override - public void handleMessage(Message message) throws MessagingException { - errorMessage.set(message); - latch.countDown(); - } - + ec.subscribe(message1 -> { + errorMessage.set(message1); + latch.countDown(); }); SubscribableChannel globalEc = binder.getApplicationContext() .getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class); - globalEc.subscribe(new MessageHandler() { - - @Override - public void handleMessage(Message message) throws MessagingException { - latch.countDown(); - } - - }); + globalEc.subscribe(message12 -> latch.countDown()); KafkaProducerMessageHandler endpoint = TestUtils.getPropertyValue(producerBinding, "lifecycle", KafkaProducerMessageHandler.class); final RuntimeException fooException = new RuntimeException("foo"); @@ -1875,11 +1953,6 @@ public abstract class AbstractKafkaBinderTests extends producerBinding.unbind(); } - @Override - protected void binderBindUnbindLatency() throws InterruptedException { - Thread.sleep(500); - } - private final class FailingInvocationCountingMessageHandler implements MessageHandler { private int invocationCount; @@ -1899,7 +1972,7 @@ public abstract class AbstractKafkaBinderTests extends @Override public void handleMessage(Message message) throws MessagingException { invocationCount++; - Long offset = message.getHeaders().get(AbstractKafkaBinderTests.this.getKafkaOffsetHeaderKey(), Long.class); + Long offset = message.getHeaders().get(KafkaBinderTests.this.getKafkaOffsetHeaderKey(), Long.class); // using the offset as key allows to ensure that we don't store duplicate // messages on retry if (!receivedMessages.containsKey(offset)) { diff --git a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java similarity index 97% rename from spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java rename to spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index 5314c8cb1..708281246 100644 --- a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -40,7 +40,7 @@ import org.springframework.kafka.support.ProducerListener; public class KafkaTestBinder extends AbstractKafkaTestBinder { @SuppressWarnings({ "rawtypes", "unchecked" }) - public KafkaTestBinder(KafkaBinderConfigurationProperties binderConfiguration) { + KafkaTestBinder(KafkaBinderConfigurationProperties binderConfiguration) { try { AdminUtilsOperation adminUtilsOperation = new KafkaAdminUtilsOperation(); KafkaTopicProvisioner provisioningProvider = diff --git a/spring-cloud-stream-binder-kafka/src/test/resources/schemas/users_v1.schema b/spring-cloud-stream-binder-kafka/src/test/resources/schemas/users_v1.schema deleted file mode 100644 index 366387be1..000000000 --- a/spring-cloud-stream-binder-kafka/src/test/resources/schemas/users_v1.schema +++ /dev/null @@ -1,8 +0,0 @@ -{"namespace": "org.springframework.cloud.stream.binder.kafka", - "type": "record", - "name": "User1", - "fields": [ - {"name": "name", "type": "string"}, - {"name": "favoriteColor", "type": "string"} - ] -} \ No newline at end of file