From 5a25a9bd098994998383942d0f371efff2483879 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 23 Dec 2015 17:14:47 -0500 Subject: [PATCH] GH-96: Catch NPE from `ClientUtils` Fixes: https://github.com/spring-projects/spring-integration-kafka/issues/96 --- .../kafka/core/DefaultConnectionFactory.java | 30 +++++++++++----- .../DefaultConnectionFactoryTests.java | 36 ++++++++++++++----- 2 files changed, 49 insertions(+), 17 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java index 4f12a3ebb9..b7aa3c039b 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java @@ -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 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(topics)), - ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString), + .fetchTopicMetadata(JavaConversions.asScalaSet(new HashSet<>(topics)), + brokers, this.configuration.getClientId(), this.configuration.getFetchMetadataTimeout(), 0)); PartitionIterable selectWithoutErrors = Iterate .partition(topicMetadataResponse.topicsMetadata(), errorlessTopicMetadataPredicate); 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 a71567d2fd..202fa92a4d 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 @@ -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)); + } + } + }