Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Deprecated Connection#getLeaders() - simplify DefaultConnectionFactory tests

Browse Source
This commit is contained in:
Marius Bogoevici
2015-10-22 15:28:12 -04:00
committed by Artem Bilan
parent 9ddc4e6d12
commit 9b49e9a673
4 changed files with 6 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java
View File

@@ -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;

1
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnection.java
View File

@@ -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());

13
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/DefaultConnectionFactoryTests.java
View File

@@ -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);
}
}

1
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/MultiBroker5ReplicatedWithBounceTests.java
View File

@@ -45,7 +45,6 @@ import org.springframework.integration.kafka.rule.KafkaEmbedded;
/**
* @author Marius Bogoevici
*/
@Ignore
public class MultiBroker5ReplicatedWithBounceTests extends AbstractMessageListenerContainerTests {
@Rule