diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java
index 4793f8a3bf..1577f40e33 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java
@@ -72,7 +72,9 @@ public interface Connection {
 	 * @param topics the topics whose partitions we query for
 	 * @return broker addresses, indexed by {@link Partition}
 	 * @throws ConsumerException the ConsumerException if any underlying error
+	 * @deprecated as of 1.3, only {@link ConnectionFactory#getLeaders(Iterable)} should be used
 	 */
+	@Deprecated
 	Result<BrokerAddress> findLeaders(String... topics) throws ConsumerException;
 
 
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnection.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnection.java
index fcc2779bd1..a1f49bb109 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnection.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnection.java
@@ -240,6 +240,7 @@ public class DefaultConnection implements Connection {
 	 * @see Connection#findLeaders(String...)
 	 */
 	@Override
+	@Deprecated
 	public Result<BrokerAddress> findLeaders(String... topics) throws ConsumerException {
 		TopicMetadataRequest topicMetadataRequest =
 				new TopicMetadataRequest(Arrays.asList(topics), createCorrelationId());
diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java
index 71e94e7a41..a71567d2fd 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java
@@ -20,6 +20,7 @@ package org.springframework.integration.kafka.listener;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
 import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 
 import java.util.List;
@@ -56,15 +57,11 @@ public class DefaultConnectionFactoryTests extends AbstractBrokerTests {
 		createTopic(TEST_TOPIC, 1, 1, 1);
 
 		BrokerAddress[] brokerAddresses = getKafkaRule().getBrokerAddresses();
-		Partition partition = new Partition(TEST_TOPIC, 0);
 		DefaultConnectionFactory connectionFactory =
 				new DefaultConnectionFactory(new BrokerAddressListConfiguration(brokerAddresses));
 		connectionFactory.afterPropertiesSet();
 		Connection connection = connectionFactory.connect(brokerAddresses[0]);
-		Result<BrokerAddress> leaders = connection.findLeaders(TEST_TOPIC);
-		assertThat(leaders.getErrors().entrySet(), empty());
-		assertThat(leaders.getResults().entrySet(), hasSize(1));
-		assertThat(leaders.getResults().get(partition), equalTo(brokerAddresses[0]));
+		assertNotNull(connection);
 	}
 
 	@Test
@@ -72,7 +69,6 @@ public class DefaultConnectionFactoryTests extends AbstractBrokerTests {
 
 		createTopic(TEST_TOPIC, 1, 1, 1);
 
-
 		Partition partition = new Partition(TEST_TOPIC, 0);
 		ZookeeperConnect zookeeperConnect = new ZookeeperConnect();
 		zookeeperConnect.setZkConnect(kafkaEmbeddedBrokerRule.getZookeeperConnectionString());
@@ -80,10 +76,7 @@ public class DefaultConnectionFactoryTests extends AbstractBrokerTests {
 				new DefaultConnectionFactory(new ZookeeperConfiguration(zookeeperConnect));
 		connectionFactory.afterPropertiesSet();
 		Connection connection = connectionFactory.connect(getKafkaRule().getBrokerAddresses()[0]);
-		Result<BrokerAddress> leaders = connection.findLeaders(TEST_TOPIC);
-		assertThat(leaders.getErrors().entrySet(), empty());
-		assertThat(leaders.getResults().entrySet(), hasSize(1));
-		assertThat(leaders.getResults().get(partition), equalTo(getKafkaRule().getBrokerAddresses()[0]));
+		assertNotNull(connection);
 	}
 
 }
diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/MultiBroker5ReplicatedWithBounceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/MultiBroker5ReplicatedWithBounceTests.java
index 7aa1fae903..1250ec0a14 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/MultiBroker5ReplicatedWithBounceTests.java
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/MultiBroker5ReplicatedWithBounceTests.java
@@ -45,7 +45,6 @@ import org.springframework.integration.kafka.rule.KafkaEmbedded;
 /**
  * @author Marius Bogoevici
  */
-@Ignore
 public class MultiBroker5ReplicatedWithBounceTests extends AbstractMessageListenerContainerTests {
 
 	@Rule