From 3c7615f7a38a63d1d9fd12c8bfcded77b447b692 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 11 Oct 2017 16:39:16 -0400 Subject: [PATCH] Test structure improvments Remove the separate test module introduced in 1.x to test different versions of Kafka. In 2.0, there is a single Kafka version that needs to be tested. Move all the tests from the test module to the main binder module. Remove the confluent schema registry integration test from the binder tests as it will be ported as a sample application. This test currently does the serialziation/deserialition twice. Fix #200 --- pom.xml | 8 - spring-cloud-stream-binder-kafka-test/.jdk8 | 0 spring-cloud-stream-binder-kafka-test/pom.xml | 135 ------- .../stream/binder/kafka/KafkaBinderTests.java | 366 ------------------ .../cloud/stream/binder/kafka/User1.java | 85 ---- spring-cloud-stream-binder-kafka/pom.xml | 23 -- ...BinderTests.java => KafkaBinderTests.java} | 355 ++++++++++------- .../stream/binder/kafka/KafkaTestBinder.java | 2 +- .../test/resources/schemas/users_v1.schema | 8 - 9 files changed, 215 insertions(+), 767 deletions(-) delete mode 100644 spring-cloud-stream-binder-kafka-test/.jdk8 delete mode 100644 spring-cloud-stream-binder-kafka-test/pom.xml delete mode 100644 spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java delete mode 100644 spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java rename spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/{AbstractKafkaBinderTests.java => KafkaBinderTests.java} (91%) rename {spring-cloud-stream-binder-kafka-test => spring-cloud-stream-binder-kafka}/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java (97%) delete mode 100644 spring-cloud-stream-binder-kafka/src/test/resources/schemas/users_v1.schema 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