From 3706405ee9bb7d256969274b65b9de7a183b3c0d Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 2 Apr 2018 16:23:53 -0400 Subject: [PATCH] GH-634: 1.1.0 Client compatibility * GH-634: 1.1.0 Client compatibility Fixes https://github.com/spring-projects/spring-kafka/issues/634 * Revert version * Fix version check * Add docs * Polishing - PR Comments --- .../kafka/test/rule/KafkaEmbedded.java | 59 +++++++++++++++++-- src/reference/asciidoc/appendix.adoc | 53 +++++++++++++++++ src/reference/asciidoc/testing.adoc | 2 + src/reference/asciidoc/whats-new.adoc | 3 + 4 files changed, 111 insertions(+), 6 deletions(-) diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java index 57384073..27f6d6a1 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java @@ -18,6 +18,9 @@ package org.springframework.kafka.test.rule; import static org.assertj.core.api.Assertions.assertThat; +import java.io.File; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -43,6 +46,7 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.security.auth.SecurityProtocol; +import org.apache.kafka.common.utils.AppInfoParser; import org.apache.kafka.common.utils.Time; import org.junit.rules.ExternalResource; @@ -85,6 +89,29 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule, Initia public static final long METADATA_PROPAGATION_TIMEOUT = 10000L; + private static final String clientVersion; + + private static final Method testUtilsCreateBrokerConfigMethod; + + static { + clientVersion = AppInfoParser.getVersion(); + if (clientVersion.startsWith("1.1.")) { + try { + testUtilsCreateBrokerConfigMethod = TestUtils.class.getDeclaredMethod("createBrokerConfig", + int.class, String.class, boolean.class, boolean.class, int.class, + scala.Option.class, scala.Option.class, scala.Option.class, + boolean.class, boolean.class, int.class, boolean.class, int.class, boolean.class, + int.class, scala.Option.class, int.class, boolean.class); + } + catch (NoSuchMethodException | SecurityException e) { + throw new RuntimeException("Failed to determine TestUtils.createBrokerConfig() method"); + } + } + else { + testUtilsCreateBrokerConfigMethod = null; + } + } + private final int count; private final boolean controlledShutdown; @@ -191,12 +218,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule, Initia ZKStringSerializer$.MODULE$); this.kafkaServers.clear(); for (int i = 0; i < this.count; i++) { - Properties brokerConfigProperties = TestUtils.createBrokerConfig(i, this.zkConnect, this.controlledShutdown, - true, this.kafkaPorts[i], - scala.Option.apply(null), - scala.Option.apply(null), - scala.Option.apply(null), - true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1); + Properties brokerConfigProperties = createBrokerProperties(i); brokerConfigProperties.setProperty(KafkaConfig.ReplicaSocketTimeoutMsProp(), "1000"); brokerConfigProperties.setProperty(KafkaConfig.ControllerSocketTimeoutMsProp(), "1000"); brokerConfigProperties.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp(), "1"); @@ -222,6 +244,31 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule, Initia System.setProperty(SPRING_EMBEDDED_ZOOKEEPER_CONNECT, getZookeeperConnectionString()); } + public Properties createBrokerProperties(int i) { + if (testUtilsCreateBrokerConfigMethod == null) { + return TestUtils.createBrokerConfig(i, this.zkConnect, this.controlledShutdown, + true, this.kafkaPorts[i], + scala.Option.apply(null), + scala.Option.apply(null), + scala.Option.apply(null), + true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1); + } + else { + try { + return (Properties) testUtilsCreateBrokerConfigMethod.invoke(null, i, this.zkConnect, + this.controlledShutdown, + true, this.kafkaPorts[i], + scala.Option.apply(null), + scala.Option.apply(null), + scala.Option.apply(null), + true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1, false); + } + catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { + throw new RuntimeException(e); + } + } + } + @Override public void destroy() throws Exception { diff --git a/src/reference/asciidoc/appendix.adoc b/src/reference/asciidoc/appendix.adoc index 1bbf38db..d72fac6d 100644 --- a/src/reference/asciidoc/appendix.adoc +++ b/src/reference/asciidoc/appendix.adoc @@ -1,3 +1,56 @@ +[[deps-for-11x]] +== Override Dependencies to use the 1.1.x kafka-clients + +When using `spring-kafka-test` (_version 2.1.x_, starting with _version 2.1.5_) with the 1.1.x `kafka-clients` jar, you will need to override certain transitive dependencies as follows: + +[source, xml] +---- + + org.springframework.kafka + spring-kafka + ${spring.kafka.version} + + + + org.springframework.kafka + spring-kafka-test + ${spring.kafka.version} + test + + + org.apache.kafka + kafka-clients + + + + + + org.apache.kafka + kafka-clients + 1.1.0 + + + + org.apache.kafka + kafka-clients + 1.1.0 + test + + + + org.apache.kafka + kafka_2.11 + 1.1.0 + + + + org.apache.kafka + kafka_2.11 + 1.1.0 + test + +---- + [[history]] == Change History diff --git a/src/reference/asciidoc/testing.adoc b/src/reference/asciidoc/testing.adoc index 9068a031..6f114773 100644 --- a/src/reference/asciidoc/testing.adoc +++ b/src/reference/asciidoc/testing.adoc @@ -5,6 +5,8 @@ The `spring-kafka-test` jar contains some useful utilities to assist with testing your applications. +NOTE: See <> if you wish to use the 1.1.x `kafka-clients` jar with _version 2.1.x_. + ==== JUnit `o.s.kafka.test.utils.KafkaTestUtils` provides some static methods to set up producer and consumer properties: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index f63c230b..deab10d6 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -4,6 +4,9 @@ This version requires the 1.0.0 `kafka-clients` or higher. +NOTE: The 1.1.x client is supported, with _version 2.1.5_, but you will need to override dependencies as described in <>. +The 1.1.x client will be supported natively in _version 2.2_. + ==== JSON Improvements The `StringJsonMessageConverter` and `JsonSerializer` now add type information in `Headers`, allowing the converter and `JsonDeserializer` to create specific types on reception, based on the message itself rather than a fixed configured type.