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

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.<SecurityProtocol>apply(null),
+					scala.Option.<File>apply(null),
+					scala.Option.<Properties>apply(null),
+					true, false, 0, false, 0, false, 0, scala.Option.<String>apply(null), 1, false);
+			}
+			catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
+				throw new RuntimeException(e);
+			}
+		}
+	}
+
 
 	@Override
 	public void destroy() throws Exception {

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]
+----
+<dependency>
+    <groupId>org.springframework.kafka</groupId>
+    <artifactId>spring-kafka</artifactId>
+    <version>${spring.kafka.version}</version>
+</dependency>
+
+<dependency>
+    <groupId>org.springframework.kafka</groupId>
+    <artifactId>spring-kafka-test</artifactId>
+    <version>${spring.kafka.version}</version>
+    <scope>test</scope>
+    <exclusions>
+        <exclusion>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </exclusion>
+    </exclusions>
+</dependency>
+
+<dependency>
+    <groupId>org.apache.kafka</groupId>
+    <artifactId>kafka-clients</artifactId>
+    <version>1.1.0</version>
+</dependency>
+
+<dependency>
+    <groupId>org.apache.kafka</groupId>
+    <artifactId>kafka-clients</artifactId>
+    <version>1.1.0</version>
+    <classifier>test</classifier>
+</dependency>
+
+<dependency>
+    <groupId>org.apache.kafka</groupId>
+    <artifactId>kafka_2.11</artifactId>
+    <version>1.1.0</version>
+</dependency>
+
+<dependency>
+    <groupId>org.apache.kafka</groupId>
+    <artifactId>kafka_2.11</artifactId>
+    <version>1.1.0</version>
+    <classifier>test</classifier>
+</dependency>
+----
+
 [[history]]
 == Change History
 

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 <<deps-for-11x>> 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:

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 <<deps-for-11x>>.
+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.