From 83eca9b7343f392b9e65feaef621a1d25f07d23a Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 6 Feb 2017 11:12:54 -0500 Subject: [PATCH] Setting default Kafka baseline to 0.10.1.1 - Starting 1.2, this is going to be the default Kafka version - Swap AdminUtilsOperation implementation for 0.9 and 0.10 (Former using reflection and straight API call for latter) - In addition to Kafka, Spring-Kafka and SI Kafka dependencies are updated as well. Separate test artifact for 0.9.0.1. - Pull out 0.9 based tests from the default binder module to this new module. Separate test artifact for 0.10.0.1. Fixes #88 Fixes #81 --- pom.xml | 9 +- .../pom.xml | 11 +- .../binder/kafka/Kafka01001BinderTests.java | 28 +++++ .../pom.xml | 78 ++++++++++++ .../binder/kafka/Kafka09BinderTests.java | 0 .../binder/kafka/Kafka09TestBinder.java | 0 .../kafka/RawModeKafka09BinderTests.java | 0 .../src/main/asciidoc/overview.adoc | 62 +++++++--- spring-cloud-stream-binder-kafka/pom.xml | 30 +++++ .../kafka/admin/AdminUtilsOperation.java | 2 +- .../admin/Kafka09AdminUtilsOperation.java | 114 +++++++++++++++-- .../admin/Kafka10AdminUtilsOperation.java | 116 ++---------------- .../binder/kafka/Kafka10BinderTests.java | 5 +- .../binder/kafka/Kafka10TestBinder.java | 0 .../cloud/stream/binder/kafka/User1.java | 0 .../test/resources/schemas/users_v1.schema | 0 16 files changed, 310 insertions(+), 145 deletions(-) rename {spring-cloud-stream-binder-kafka-0.10-test => spring-cloud-stream-binder-kafka-0.10.0.1-test}/pom.xml (89%) create mode 100644 spring-cloud-stream-binder-kafka-0.10.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka01001BinderTests.java create mode 100644 spring-cloud-stream-binder-kafka-0.9.0.1-test/pom.xml rename {spring-cloud-stream-binder-kafka => spring-cloud-stream-binder-kafka-0.9.0.1-test}/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java (100%) rename {spring-cloud-stream-binder-kafka => spring-cloud-stream-binder-kafka-0.9.0.1-test}/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java (100%) rename {spring-cloud-stream-binder-kafka => spring-cloud-stream-binder-kafka-0.9.0.1-test}/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java (100%) rename {spring-cloud-stream-binder-kafka-0.10-test => spring-cloud-stream-binder-kafka}/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java (98%) rename {spring-cloud-stream-binder-kafka-0.10-test => spring-cloud-stream-binder-kafka}/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java (100%) rename {spring-cloud-stream-binder-kafka-0.10-test => spring-cloud-stream-binder-kafka}/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java (100%) rename {spring-cloud-stream-binder-kafka-0.10-test => spring-cloud-stream-binder-kafka}/src/test/resources/schemas/users_v1.schema (100%) diff --git a/pom.xml b/pom.xml index 57b1ffbea..04e877bd0 100644 --- a/pom.xml +++ b/pom.xml @@ -12,16 +12,17 @@ 1.7 - 0.9.0.1 - 1.0.5.RELEASE - 2.0.1.RELEASE + 0.10.1.1 + 1.1.2.RELEASE + 2.1.0.RELEASE 1.2.0.BUILD-SNAPSHOT spring-cloud-stream-binder-kafka spring-cloud-starter-stream-kafka spring-cloud-stream-binder-kafka-docs - spring-cloud-stream-binder-kafka-0.10-test + spring-cloud-stream-binder-kafka-0.9.0.1-test + spring-cloud-stream-binder-kafka-0.10.0.1-test diff --git a/spring-cloud-stream-binder-kafka-0.10-test/pom.xml b/spring-cloud-stream-binder-kafka-0.10.0.1-test/pom.xml similarity index 89% rename from spring-cloud-stream-binder-kafka-0.10-test/pom.xml rename to spring-cloud-stream-binder-kafka-0.10.0.1-test/pom.xml index ad5717789..8c7505f27 100644 --- a/spring-cloud-stream-binder-kafka-0.10-test/pom.xml +++ b/spring-cloud-stream-binder-kafka-0.10.0.1-test/pom.xml @@ -6,8 +6,8 @@ spring-cloud-stream-binder-kafka-parent 1.2.0.BUILD-SNAPSHOT - spring-cloud-stream-binder-kafka-0.10-test - Spring Cloud Stream Kafka Binder 0.10 Tests + spring-cloud-stream-binder-kafka-0.10.0.1-test + Spring Cloud Stream Kafka Binder 0.10.0.1 Tests http://projects.spring.io/spring-cloud Pivotal Software, Inc. @@ -16,11 +16,11 @@ ${basedir}/../.. - 0.10.0.0 - 1.1.1.RELEASE + 0.10.0.1 + 1.1.2.RELEASE 2.1.0.RELEASE @@ -100,5 +100,4 @@ - diff --git a/spring-cloud-stream-binder-kafka-0.10.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka01001BinderTests.java b/spring-cloud-stream-binder-kafka-0.10.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka01001BinderTests.java new file mode 100644 index 000000000..67050befa --- /dev/null +++ b/spring-cloud-stream-binder-kafka-0.10.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka01001BinderTests.java @@ -0,0 +1,28 @@ +/* + * 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; + +/** + * Integration tests for the {@link KafkaMessageChannelBinder}. + * + * This test specifically tests for the 0.10.0.1 version of Kafka. + * + * @author Soby Chacko + */ +public class Kafka01001BinderTests extends Kafka10BinderTests { + +} diff --git a/spring-cloud-stream-binder-kafka-0.9.0.1-test/pom.xml b/spring-cloud-stream-binder-kafka-0.9.0.1-test/pom.xml new file mode 100644 index 000000000..9a3aca98f --- /dev/null +++ b/spring-cloud-stream-binder-kafka-0.9.0.1-test/pom.xml @@ -0,0 +1,78 @@ + + + 4.0.0 + + org.springframework.cloud + spring-cloud-stream-binder-kafka-parent + 1.2.0.BUILD-SNAPSHOT + + spring-cloud-stream-binder-kafka-0.9.0.1-test + Spring Cloud Stream Kafka Binder 0.9.0.1 Tests + http://projects.spring.io/spring-cloud + + Pivotal Software, Inc. + http://www.spring.io + + + ${basedir}/../.. + + 0.9.0.1 + 1.0.5.RELEASE + 2.0.1.RELEASE + + + + + org.springframework.cloud + spring-cloud-stream-binder-kafka + ${project.version} + 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.cloud + spring-cloud-stream-binder-kafka + ${project.version} + test-jar + test + + + org.springframework.cloud + spring-cloud-stream-binder-test + test + + + + diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java b/spring-cloud-stream-binder-kafka-0.9.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java similarity index 100% rename from spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java rename to spring-cloud-stream-binder-kafka-0.9.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java b/spring-cloud-stream-binder-kafka-0.9.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java similarity index 100% rename from spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java rename to spring-cloud-stream-binder-kafka-0.9.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java b/spring-cloud-stream-binder-kafka-0.9.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java similarity index 100% rename from spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java rename to spring-cloud-stream-binder-kafka-0.9.0.1-test/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index 5c435d6f3..89cc90fa6 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -328,26 +328,18 @@ In secure environments, we strongly recommend creating topics and managing ACLs ==== Using the binder with Apache Kafka 0.10 -The binder also supports connecting to Kafka 0.10 brokers. -In order to support this, when you create the project that contains your application, include `spring-cloud-starter-stream-kafka` as you normally would do for 0.9 based applications. -Then add these dependencies at the top of the `` section in the pom.xml file to override the Apache Kafka, Spring Kafka, and Spring Integration Kafka with 0.10-compatible versions as in the following example: +The default Kafka support in Spring Cloud Stream Kafka binder is for Kafka version 0.10.1.1. The binder also supports connecting to other 0.10 based versions and 0.9 clients. +In order to do this, when you create the project that contains your application, include `spring-cloud-starter-stream-kafka` as you normally would do for the default binder. +Then add these dependencies at the top of the `` section in the pom.xml file to override the dependencies. + +Here is an example for downgrading your application to 0.10.0.1. Since it is still on the 0.10 line, the default `spring-kafka` and `spring-integration-kafka` versions can be retained. [source,xml] ---- - - org.springframework.kafka - spring-kafka - 1.1.1.RELEASE - - - org.springframework.integration - spring-integration-kafka - 2.1.0.RELEASE - org.apache.kafka kafka_2.11 - 0.10.0.0 + 0.10.0.1 org.slf4j @@ -355,6 +347,44 @@ Then add these dependencies at the top of the `` section in the po + + org.apache.kafka + kafka-clients + 0.10.0.1 + +---- + +Here is another example of using 0.9.0.1 version. + +[source,xml] +---- + + org.springframework.kafka + spring-kafka + 1.0.5.RELEASE + + + org.springframework.integration + spring-integration-kafka + 2.0.1.RELEASE + + + org.apache.kafka + kafka_2.11 + 0.9.0.1 + + + org.slf4j + slf4j-log4j12 + + + + + org.apache.kafka + kafka-clients + 0.9.0.1 + + ---- [NOTE] @@ -369,8 +399,8 @@ For best results, we recommend using the most recent 0.10-compatible versions of The Apache Kafka Binder uses the administrative utilities which are part of the Apache Kafka server library to create and reconfigure topics. If the inclusion of the Apache Kafka server library and its dependencies is not necessary at runtime because the application will rely on the topics being configured administratively, the Kafka binder allows for Apache Kafka server dependency to be excluded from the application. -If you use Kafka 10 dependencies as advised above, all you have to do is not to include the kafka broker dependency. -If you use Kafka 0.9, then ensure that you exclude the kafka broker jar from the `spring-cloud-starter-stream-kafka` dependency as following. +If you use non default versions for Kafka dependencies as advised above, all you have to do is not to include the kafka broker dependency. +If you use the default Kafka version, then ensure that you exclude the kafka broker jar from the `spring-cloud-starter-stream-kafka` dependency as following. [source,xml] ---- diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml index fa8a610cb..deafb7966 100644 --- a/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binder-kafka/pom.xml @@ -71,8 +71,38 @@ test 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.1.2 + test + + + io.confluent + kafka-schema-registry + 3.1.2 + test + + + + confluent + http://packages.confluent.io/maven/ + + + diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java index b36ce166e..4cb2b8bff 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java @@ -24,7 +24,7 @@ import kafka.utils.ZkUtils; * API around {@link kafka.admin.AdminUtils} to support * various versions of Kafka brokers. * - * Note: Implementations that support Kafka brokers other than 0.9, need to use + * Note: Implementations that support Kafka brokers other than 0.10, need to use * a possible strategy that involves reflection around {@link kafka.admin.AdminUtils}. * * @author Soby Chacko diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java index bbec0f6b3..dfacbf3bc 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java @@ -16,36 +16,130 @@ package org.springframework.cloud.stream.binder.kafka.admin; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.util.Properties; -import kafka.admin.AdminUtils; -import kafka.api.TopicMetadata; import kafka.utils.ZkUtils; +import org.springframework.util.ClassUtils; +import org.springframework.util.ReflectionUtils; + /** * @author Soby Chacko */ public class Kafka09AdminUtilsOperation implements AdminUtilsOperation { + private static Class ADMIN_UTIL_CLASS; + + static { + try { + ADMIN_UTIL_CLASS = ClassUtils.forName("kafka.admin.AdminUtils", null); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("AdminUtils class not found", e); + } + } + public void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions, String replicaAssignmentStr, boolean checkBrokerAvailable) { - AdminUtils.addPartitions(zkUtils, topic, numPartitions, - replicaAssignmentStr, checkBrokerAvailable); + try { + Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods(); + Method addPartitions = null; + for (Method m : declaredMethods) { + if (m.getName().equals("addPartitions")) { + addPartitions = m; + } + } + if (addPartitions != null) { + addPartitions.invoke(null, zkUtils, topic, numPartitions, + replicaAssignmentStr, checkBrokerAvailable); + } + else { + throw new InvocationTargetException( + new RuntimeException("method not found")); + } + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } + } public short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils) { - TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); - return topicMetadata.errorCode(); + try { + Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class); + Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils); + Class topicMetadataClass = ClassUtils.forName("kafka.api.TopicMetadata", null); + Method errorCodeMethod = ReflectionUtils.findMethod(topicMetadataClass, "errorCode"); + return (short) errorCodeMethod.invoke(result); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("AdminUtils class not found", e); + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } + return 0; } + @SuppressWarnings("unchecked") public int partitionSize(String topic, ZkUtils zkUtils) { - TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); - return topicMetadata.partitionsMetadata().size(); + try { + Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class); + Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils); + Class topicMetadataClass = ClassUtils.forName("kafka.api.TopicMetadata", null); + + Method partitionsMetadata = ReflectionUtils.findMethod(topicMetadataClass, "partitionsMetadata"); + scala.collection.Seq partitionSize = + (scala.collection.Seq)partitionsMetadata.invoke(result); + + return partitionSize.size(); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("AdminUtils class not found", e); + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } + return 0; + } public void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) { - AdminUtils.createTopic(zkUtils, topic, partitions, replicationFactor, - topicConfig); + try { + Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods(); + Method createTopic = null; + for (Method m : declaredMethods) { + if (m.getName().equals("createTopic")) { + createTopic = m; + break; + } + } + if (createTopic != null) { + createTopic.invoke(null, zkUtils, topic, partitions, + replicationFactor, topicConfig); + } + else { + throw new InvocationTargetException( + new RuntimeException("method not found")); + } + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java index 8bc5e47e9..026ffe6a6 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java @@ -16,135 +16,39 @@ package org.springframework.cloud.stream.binder.kafka.admin; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.util.List; import java.util.Properties; -import kafka.api.PartitionMetadata; +import kafka.admin.AdminUtils; import kafka.utils.ZkUtils; - -import org.springframework.util.ClassUtils; -import org.springframework.util.ReflectionUtils; +import org.apache.kafka.common.requests.MetadataResponse; /** * @author Soby Chacko */ public class Kafka10AdminUtilsOperation implements AdminUtilsOperation { - private static Class ADMIN_UTIL_CLASS; - - static { - try { - ADMIN_UTIL_CLASS = ClassUtils.forName("kafka.admin.AdminUtils", null); - } - catch (ClassNotFoundException e) { - throw new IllegalStateException("AdminUtils class not found", e); - } - } - public void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions, String replicaAssignmentStr, boolean checkBrokerAvailable) { - try { - Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods(); - Method addPartitions = null; - for (Method m : declaredMethods) { - if (m.getName().equals("addPartitions")) { - addPartitions = m; - } - } - - if (addPartitions != null) { - addPartitions.invoke(null, zkUtils, topic, numPartitions, - replicaAssignmentStr, checkBrokerAvailable, null); - } - else { - throw new InvocationTargetException( - new RuntimeException("method not found")); - } - } - catch (InvocationTargetException e) { - ReflectionUtils.handleInvocationTargetException(e); - } - catch (IllegalAccessException e) { - ReflectionUtils.handleReflectionException(e); - } + AdminUtils.addPartitions(zkUtils, topic, numPartitions, replicaAssignmentStr, checkBrokerAvailable, null); } public short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils) { - try { - Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class); - - Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils); - Class topicMetadataClass = ClassUtils.forName("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata", null); - - Method errorCodeMethod = ReflectionUtils.findMethod(topicMetadataClass, "error"); - Object obj = errorCodeMethod.invoke(result); - Method code = ReflectionUtils.findMethod(obj.getClass(), "code"); - - return (short) code.invoke(obj); - } - catch (ClassNotFoundException e) { - throw new IllegalStateException("AdminUtils class not found", e); - } - catch (InvocationTargetException e) { - ReflectionUtils.handleInvocationTargetException(e); - } - catch (IllegalAccessException e) { - ReflectionUtils.handleReflectionException(e); - } - return 0; + MetadataResponse.TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); + return topicMetadata.error().code(); } @SuppressWarnings("unchecked") public int partitionSize(String topic, ZkUtils zkUtils) { - try { - Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class); - Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils); - Class topicMetadataClass = ClassUtils.forName("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata", null); - Method partitionsMetadata = ReflectionUtils.findMethod(topicMetadataClass, "partitionMetadata"); - List foo = (List) partitionsMetadata.invoke(result); - return foo.size(); - } - catch (ClassNotFoundException e) { - throw new IllegalStateException("AdminUtils class not found", e); - } - catch (InvocationTargetException e) { - ReflectionUtils.handleInvocationTargetException(e); - } - catch (IllegalAccessException e) { - ReflectionUtils.handleReflectionException(e); - } - return 0; + MetadataResponse.TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); + return topicMetadata.partitionMetadata().size(); } public void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) { - try { - Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods(); - Method createTopic = null; - for (Method m : declaredMethods) { - if (m.getName().equals("createTopic") && m.getParameterTypes()[m.getParameterTypes().length - 1].getName().endsWith("RackAwareMode")) { - createTopic = m; - break; - } - } - if (createTopic != null) { - createTopic.invoke(null, zkUtils, topic, partitions, - replicationFactor, topicConfig, null); - } - else { - throw new InvocationTargetException( - new RuntimeException("method not found")); - } - } - catch (InvocationTargetException e) { - ReflectionUtils.handleInvocationTargetException(e); - } - catch (IllegalAccessException e) { - ReflectionUtils.handleReflectionException(e); - } + + AdminUtils.createTopic(zkUtils, topic, partitions, replicationFactor, + topicConfig, null); } } diff --git a/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java similarity index 98% rename from spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java rename to spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java index 798fe6a8c..9556faf43 100644 --- a/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java @@ -31,6 +31,7 @@ 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; @@ -237,8 +238,8 @@ public class Kafka10BinderTests extends KafkaBinderTests { assertThat(inbound).isNotNull(); assertTrue(message.getPayload() instanceof User1); User1 receivedUser = (User1) message.getPayload(); - assertThat(receivedUser.getName()).isEqualTo(userName1); - assertThat(receivedUser.getFavoriteColor()).isEqualTo(favColor1); + 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-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java similarity index 100% rename from spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java rename to spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java diff --git a/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java similarity index 100% rename from spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java rename to spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/User1.java diff --git a/spring-cloud-stream-binder-kafka-0.10-test/src/test/resources/schemas/users_v1.schema b/spring-cloud-stream-binder-kafka/src/test/resources/schemas/users_v1.schema similarity index 100% rename from spring-cloud-stream-binder-kafka-0.10-test/src/test/resources/schemas/users_v1.schema rename to spring-cloud-stream-binder-kafka/src/test/resources/schemas/users_v1.schema