diff --git a/spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml b/spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml deleted file mode 100644 index 37dd0ec6b..000000000 --- a/spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml +++ /dev/null @@ -1,126 +0,0 @@ - - - 4.0.0 - - org.springframework.cloud - spring-cloud-stream-binder-kafka-parent - 1.3.1.BUILD-SNAPSHOT - - spring-cloud-stream-binder-kafka-0.10.2-test - Spring Cloud Stream Kafka Binder 0.10.2 Tests - http://projects.spring.io/spring-cloud - - Pivotal Software, Inc. - http://www.spring.io - - - ${basedir}/../.. - 0.10.2.1 - 1.2.2.RELEASE - - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka-core - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - test - - - org.springframework.kafka - spring-kafka - test - - - org.apache.kafka - kafka_2.11 - test - - - org.slf4j - slf4j-log4j12 - - - - - 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-kafka - ${project.version} - test-jar - test - - - 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.2.2 - test - - - io.confluent - kafka-schema-registry - 3.2.2 - test - - - - - - confluent - http://packages.confluent.io/maven/ - - - - - - - org.apache.maven.plugins - maven-jar-plugin - 3.0.2 - - - - test-jar - - - - - - - - diff --git a/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java b/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java deleted file mode 100644 index 3ff966c75..000000000 --- a/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java +++ /dev/null @@ -1,86 +0,0 @@ -/* - * Copyright 2015-2017 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 org.springframework.cloud.stream.binder.ExtendedConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; -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.provisioning.KafkaTopicProvisioner; -import org.springframework.cloud.stream.provisioning.ConsumerDestination; -import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.context.annotation.Configuration; -import org.springframework.integration.config.EnableIntegration; -import org.springframework.kafka.support.LoggingProducerListener; -import org.springframework.kafka.support.ProducerListener; - -/** - * Test support class for {@link KafkaMessageChannelBinder}. - * @author Eric Bottard - * @author Marius Bogoevici - * @author David Turanski - * @author Gary Russell - * @author Soby Chacko - */ -public class Kafka10TestBinder extends AbstractKafkaTestBinder { - - @SuppressWarnings({ "rawtypes", "unchecked" }) - public Kafka10TestBinder(KafkaBinderConfigurationProperties binderConfiguration) { - try { - AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation(); - KafkaTopicProvisioner provisioningProvider = - new KafkaTopicProvisioner(binderConfiguration, adminUtilsOperation); - provisioningProvider.afterPropertiesSet(); - - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration, - provisioningProvider) { - - /* - * Some tests use multiple instance indexes for the same topic; we need to make - * the error infrastructure beans unique. - */ - @Override - protected String errorsBaseName(ConsumerDestination destination, String group, - ExtendedConsumerProperties consumerProperties) { - return super.errorsBaseName(destination, group, consumerProperties) + "-" - + consumerProperties.getInstanceIndex(); - } - - }; - - binder.setCodec(AbstractKafkaTestBinder.getCodec()); - ProducerListener producerListener = new LoggingProducerListener(); - binder.setProducerListener(producerListener); - AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class); - setApplicationContext(context); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - this.setBinder(binder); - } - catch (Exception e) { - throw new RuntimeException(e); - } - } - - @Configuration - @EnableIntegration - static class Config { - - } - -} diff --git a/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_2_BinderTests.java b/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_2_BinderTests.java deleted file mode 100644 index 605af81db..000000000 --- a/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_2_BinderTests.java +++ /dev/null @@ -1,241 +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.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 static org.junit.Assert.assertTrue; - -/** - * Integration tests for the {@link KafkaMessageChannelBinder}. - * - * This test specifically tests for the 0.10.2.x version of Kafka. - * - * @author Eric Bottard - * @author Marius Bogoevici - * @author Mark Fisher - * @author Ilayaperumal Gopinathan - */ -public class Kafka_0_10_2_BinderTests extends KafkaBinderTests { - - private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); - - @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); - - private Kafka10TestBinder binder; - - private Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation(); - - @Override - protected void binderBindUnbindLatency() throws InterruptedException { - Thread.sleep(500); - } - - @Override - protected Kafka10TestBinder getBinder() { - if (binder == null) { - KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binder = new Kafka10TestBinder(binderConfiguration); - } - return binder; - } - - 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 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 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); - } - - @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(); - } -} diff --git a/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java b/spring-cloud-stream-binder-kafka-0.10.2-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-0.10.2-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); - } - } -}