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 1870c56676..37daf9b499 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 @@ -162,9 +162,11 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac PartitionIterable selectWithoutErrors = Iterate.partition(topicMetadataResponse.topicsMetadata(), errorlessTopicMetadataPredicate); this.metadataCacheHolder.set(this.metadataCacheHolder.get().merge(selectWithoutErrors.getSelected())); - for (TopicMetadata topicMetadata : selectWithoutErrors.getRejected()) { - log.error(String.format("No metadata could be retrieved for '%s'", topicMetadata.topic()), - ErrorMapping.exceptionFor(topicMetadata.errorCode())); + if (log.isInfoEnabled()) { + for (TopicMetadata topicMetadata : selectWithoutErrors.getRejected()) { + log.info(String.format("No metadata could be retrieved for '%s'", topicMetadata.topic()), + ErrorMapping.exceptionFor(topicMetadata.errorCode())); + } } } finally { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/Testing.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/Testing.java new file mode 100644 index 0000000000..5a94db23df --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/Testing.java @@ -0,0 +1,116 @@ +/* + * Copyright 2014 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.kafka.listener; + +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; + +import org.junit.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.kafka.core.BrokerAddress; +import org.springframework.integration.kafka.core.BrokerAddressListConfiguration; +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.core.DefaultConnectionFactory; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.listener.AbstractMessageListenerContainerTest.KeyedMessageWithOffset; +import org.springframework.integration.kafka.serializer.common.StringDecoder; +import org.springframework.integration.kafka.support.KafkaHeaders; +import org.springframework.integration.metadata.PropertiesPersistingMetadataStore; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; + +import com.gs.collections.api.multimap.list.MutableListMultimap; +import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; + +/** + * @author Gary Russell + * @since 4.1 + * + */ +public class Testing { + + @Test + public void test() throws Exception { + ArrayList readPartitions = new ArrayList(); + for (int i = 0; i < 5; i++) { + readPartitions.add(new Partition("test-topic", i)); + } + ConnectionFactory kafkaBrokerConnectionFactory = getKafkaBrokerConnectionFactory(); + final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer( + kafkaBrokerConnectionFactory, readPartitions.toArray(new Partition[readPartitions.size()])); + kafkaMessageListenerContainer.setMaxFetch(100); + kafkaMessageListenerContainer.setConcurrency(2); + PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore(); + metadataStore.afterPropertiesSet(); + MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(kafkaBrokerConnectionFactory); + offsetManager.setMetadataStore(metadataStore); + kafkaMessageListenerContainer.setOffsetManager(offsetManager); + + int expectedMessageCount = 100; + + final MutableListMultimap receivedData = new SynchronizedPutFastListMultimap(); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + + KafkaMessageDrivenChannelAdapter kafkaInboundChannelAdapter = new KafkaMessageDrivenChannelAdapter(kafkaMessageListenerContainer); + kafkaInboundChannelAdapter.setBeanFactory(mock(BeanFactory.class)); + + StringDecoder decoder = new StringDecoder(); + kafkaInboundChannelAdapter.setKeyDecoder(decoder); + kafkaInboundChannelAdapter.setPayloadDecoder(decoder); + kafkaInboundChannelAdapter.setOutputChannel(new MessageChannel() { + @Override + public boolean send(Message message) { + latch.countDown(); + System.out.println(message); + return receivedData.put( + (Integer)message.getHeaders().get(KafkaHeaders.PARTITION_ID), + new KeyedMessageWithOffset( + (String)message.getHeaders().get(KafkaHeaders.MESSAGE_KEY), + (String)message.getPayload(), + (Long)message.getHeaders().get(KafkaHeaders.OFFSET), + Thread.currentThread().getName(), + (Integer)message.getHeaders().get(KafkaHeaders.PARTITION_ID))); + } + + + @Override + public boolean send(Message message, long timeout) { + return send(message); + } + }); + + kafkaInboundChannelAdapter.afterPropertiesSet(); + kafkaInboundChannelAdapter.start(); + + Thread.sleep(100000); + + kafkaInboundChannelAdapter.stop(); + metadataStore.destroy(); + } + + public ConnectionFactory getKafkaBrokerConnectionFactory() throws Exception { + DefaultConnectionFactory kafkaBrokerConnectionFactory = new DefaultConnectionFactory( + new BrokerAddressListConfiguration(Collections.singletonList(BrokerAddress.fromAddress("localhost:9092")))); + kafkaBrokerConnectionFactory.afterPropertiesSet(); + return kafkaBrokerConnectionFactory; + } + +}