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

GH-96: Catch NPE from ClientUtils

Browse Source 
Fixes: https://github.com/spring-projects/spring-integration-kafka/issues/96
This commit is contained in:
Artem Bilan
2015-12-23 17:14:47 -05:00
committed by Artem Bilan
parent 079fa03563
commit 5a25a9bd09
2 changed files with 49 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-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.
@@ -24,6 +24,13 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import com.gs.collections.api.block.function.Function;
import com.gs.collections.api.block.predicate.Predicate;
import com.gs.collections.api.partition.PartitionIterable;
@@ -31,22 +38,20 @@ import com.gs.collections.impl.block.factory.Functions;
import com.gs.collections.impl.map.mutable.UnifiedMap;
import com.gs.collections.impl.utility.Iterate;
import com.gs.collections.impl.utility.ListIterate;
import kafka.client.ClientUtils$;
import kafka.cluster.Broker;
import kafka.common.ErrorMapping;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.TopicMetadataResponse;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import scala.collection.JavaConversions;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import scala.collection.Seq;
/**
* Default implementation of {@link ConnectionFactory}
*
* @author Marius Bogoevici
* @author Artem Bilan
*/
public class DefaultConnectionFactory implements InitializingBean, ConnectionFactory, DisposableBean {
@@ -169,9 +174,16 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac
this.lock.writeLock().lock();
String brokerAddressesAsString = ListIterate
.collect(this.configuration.getBrokerAddresses(), Functions.getToString()).makeString(",");
Seq<Broker> brokers = null;
try {
brokers = ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString);
}
catch (Exception e) {
throw new IllegalStateException("Can not parse Kafka Brokers for: [" + brokerAddressesAsString + "]", e);
}
TopicMetadataResponse topicMetadataResponse = new TopicMetadataResponse(ClientUtils$.MODULE$
.fetchTopicMetadata(JavaConversions.asScalaSet(new HashSet<String>(topics)),
ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString),
.fetchTopicMetadata(JavaConversions.asScalaSet(new HashSet<>(topics)),
brokers,
this.configuration.getClientId(), this.configuration.getFetchMetadataTimeout(), 0));
PartitionIterable<TopicMetadata> selectWithoutErrors = Iterate
.partition(topicMetadataResponse.topicsMetadata(), errorlessTopicMetadataPredicate);

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

@@ -17,29 +17,31 @@
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.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.List;
import java.util.Collections;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.integration.kafka.core.BrokerAddress;
import org.springframework.integration.kafka.core.BrokerAddressListConfiguration;
import org.springframework.integration.kafka.core.Configuration;
import org.springframework.integration.kafka.core.Connection;
import org.springframework.integration.kafka.core.DefaultConnectionFactory;
import org.springframework.integration.kafka.core.BrokerAddressListConfiguration;
import org.springframework.integration.kafka.core.Result;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.core.ZookeeperConfiguration;
import org.springframework.integration.kafka.rule.KafkaEmbedded;
import org.springframework.integration.kafka.support.ZookeeperConnect;
/**
* @author Marius Bogoevici
* @author Artem Bilan
*/
public class DefaultConnectionFactoryTests extends AbstractBrokerTests {
@@ -69,7 +71,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());
DefaultConnectionFactory connectionFactory =
@@ -79,4 +80,23 @@ public class DefaultConnectionFactoryTests extends AbstractBrokerTests {
assertNotNull(connection);
}
@Test
public void testNpeFromClientUtils() throws Exception {
createTopic(TEST_TOPIC, 1, 1, 1);
Configuration configuration = mock(Configuration.class);
when(configuration.getBrokerAddresses())
.thenReturn(Collections.singletonList(BrokerAddress.fromAddress("localhost,:9092")));
DefaultConnectionFactory connectionFactory = new DefaultConnectionFactory(configuration);
connectionFactory.afterPropertiesSet();
try {
connectionFactory.refreshMetadata(Collections.singletonList(TEST_TOPIC));
fail("IllegalStateException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("Can not parse Kafka Brokers for"));
assertThat(e.getCause(), instanceOf(NullPointerException.class));
}
}
}