diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java index 263ea17286..a52707c66f 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java @@ -35,6 +35,12 @@ public interface ConnectionFactory { */ Connection connect(BrokerAddress brokerAddress); + /** + * Close the connection to the broker + * @param brokerAddress brokerAddress + */ + void disconnect(BrokerAddress brokerAddress); + /** * Retrieve the leaders for a set of partitions. * @param partitions whose leaders are queried 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 c11a15aeec..fed875d8ab 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 @@ -24,6 +24,13 @@ import java.util.Map; import java.util.Random; import java.util.concurrent.atomic.AtomicInteger; +import com.gs.collections.api.block.function.Function; +import com.gs.collections.api.block.function.Function2; +import com.gs.collections.api.tuple.Pair; +import com.gs.collections.impl.list.mutable.FastList; +import com.gs.collections.impl.tuple.Tuples; +import com.gs.collections.impl.utility.LazyIterate; +import com.gs.collections.impl.utility.MapIterate; import kafka.api.FetchRequestBuilder; import kafka.api.PartitionOffsetRequestInfo; import kafka.cluster.Broker; @@ -45,18 +52,10 @@ import kafka.javaapi.TopicMetadataResponse; import kafka.javaapi.consumer.SimpleConsumer; import kafka.javaapi.message.ByteBufferMessageSet; import kafka.message.MessageAndOffset; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.util.Assert; -import com.gs.collections.api.block.function.Function; -import com.gs.collections.api.block.function.Function2; -import com.gs.collections.api.tuple.Pair; -import com.gs.collections.impl.list.mutable.FastList; -import com.gs.collections.impl.tuple.Tuples; -import com.gs.collections.impl.utility.LazyIterate; -import com.gs.collections.impl.utility.MapIterate; +import org.springframework.util.Assert; /** * A connection to a Kafka broker. @@ -123,8 +122,8 @@ public class DefaultConnection implements Connection { ResultBuilder resultBuilder = new ResultBuilder(); for (final FetchRequest request : requests) { Partition partition = request.getPartition(); - if (log.isDebugEnabled()) { - log.debug("Reading from " + partition + "@" + request.getOffset()); + if (log.isTraceEnabled()) { + log.trace("Reading from " + partition + "@" + request.getOffset()); } short errorCode = fetchResponse.errorCode(partition.getTopic(), partition.getId()); if (ErrorMapping.NoError() == errorCode) { 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 98f5c084e1..4f12a3ebb9 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 @@ -24,13 +24,6 @@ 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; @@ -38,13 +31,18 @@ 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.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; + /** * Default implementation of {@link ConnectionFactory} * @@ -60,8 +58,8 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac private final Configuration configuration; - private final AtomicReference metadataCacheHolder = - new AtomicReference(new MetadataCache(Collections.emptySet())); + private final AtomicReference metadataCacheHolder = new AtomicReference( + new MetadataCache(Collections.emptySet())); private final ReadWriteLock lock = new ReentrantReadWriteLock(); @@ -169,20 +167,14 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac public void refreshMetadata(Collection topics) { try { this.lock.writeLock().lock(); - for (Connection connection : this.kafkaBrokersCache) { - connection.close(); - } - String brokerAddressesAsString = - ListIterate.collect(this.configuration.getBrokerAddresses(), Functions.getToString()) - .makeString(","); - TopicMetadataResponse topicMetadataResponse = - new TopicMetadataResponse( - ClientUtils$.MODULE$.fetchTopicMetadata( - JavaConversions.asScalaSet(new HashSet(topics)), - ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString), - this.configuration.getClientId(), this.configuration.getFetchMetadataTimeout(), 0)); - PartitionIterable selectWithoutErrors = Iterate.partition(topicMetadataResponse.topicsMetadata(), - errorlessTopicMetadataPredicate); + String brokerAddressesAsString = ListIterate + .collect(this.configuration.getBrokerAddresses(), Functions.getToString()).makeString(","); + TopicMetadataResponse topicMetadataResponse = new TopicMetadataResponse(ClientUtils$.MODULE$ + .fetchTopicMetadata(JavaConversions.asScalaSet(new HashSet(topics)), + ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString), + this.configuration.getClientId(), this.configuration.getFetchMetadataTimeout(), 0)); + PartitionIterable selectWithoutErrors = Iterate + .partition(topicMetadataResponse.topicsMetadata(), errorlessTopicMetadataPredicate); this.metadataCacheHolder.set(this.metadataCacheHolder.get().merge(selectWithoutErrors.getSelected())); if (log.isInfoEnabled()) { for (TopicMetadata topicMetadata : selectWithoutErrors.getRejected()) { @@ -196,6 +188,20 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac } } + @Override + public void disconnect(BrokerAddress brokerAddress) { + try { + this.lock.writeLock().lock(); + Connection connection = this.kafkaBrokersCache.get(brokerAddress); + if (connection != null) { + connection.close(); + } + } + finally { + this.lock.writeLock().unlock(); + } + } + /** * @see ConnectionFactory#getPartitions(String) */ diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java index 455d9e4c2e..68f789aad1 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java @@ -21,16 +21,16 @@ import java.util.Collection; import java.util.List; import java.util.concurrent.Executor; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.integration.kafka.core.KafkaMessage; -import org.springframework.integration.kafka.core.Partition; -import org.springframework.util.Assert; - import com.gs.collections.api.block.procedure.Procedure; import com.gs.collections.api.block.procedure.Procedure2; import com.gs.collections.api.map.MutableMap; import com.gs.collections.impl.factory.Maps; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.kafka.core.KafkaMessage; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.util.Assert; /** * Dispatches {@link KafkaMessage}s to a {@link MessageListener}. Messages may be @@ -71,14 +71,12 @@ class ConcurrentMessageListenerDispatcher { private boolean autoCommitOnError; public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler, - Collection partitions, OffsetManager offsetManager, - int consumers, int queueSize, Executor taskExecutor, - boolean autoCommitOnError) { - Assert.isTrue - (delegateListener instanceof MessageListener - || delegateListener instanceof AcknowledgingMessageListener, - "Either a " + MessageListener.class.getName() + " or a " - + AcknowledgingMessageListener.class.getName() + " must be provided"); + Collection partitions, OffsetManager offsetManager, int consumers, int queueSize, + Executor taskExecutor, boolean autoCommitOnError) { + Assert.isTrue( + delegateListener instanceof MessageListener || delegateListener instanceof AcknowledgingMessageListener, + "Either a " + MessageListener.class.getName() + " or a " + AcknowledgingMessageListener.class.getName() + + " must be provided"); Assert.notEmpty(partitions, "A set of partitions must be provided"); Assert.isTrue(consumers <= partitions.size(), "The number of consumers must be smaller or equal to the number of partitions"); @@ -120,17 +118,16 @@ class ConcurrentMessageListenerDispatcher { private void initializeAndStartDispatching() { // allocate delegate instances index them - List delegateList = new ArrayList(consumers); + List delegateList = new ArrayList<>(consumers); for (int i = 0; i < consumers; i++) { - QueueingMessageListenerInvoker queueingMessageListenerInvoker = - new QueueingMessageListenerInvoker(queueSize, offsetManager, delegateListener, errorHandler, - taskExecutor, autoCommitOnError); + QueueingMessageListenerInvoker queueingMessageListenerInvoker = new QueueingMessageListenerInvoker( + queueSize, offsetManager, delegateListener, errorHandler, taskExecutor, autoCommitOnError); delegateList.add(queueingMessageListenerInvoker); } // evenly distribute partitions across delegates - delegates = Maps.mutable.of(); int i = 0; + delegates = Maps.mutable.of(); for (Partition partition : partitions) { delegates.put(partition, delegateList.get((i++) % consumers)); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java index 3aa8f0ea0e..b8edaa7aab 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java @@ -16,25 +16,39 @@ package org.springframework.integration.kafka.listener; -import static com.gs.collections.impl.utility.ArrayIterate.flatCollect; -import static com.gs.collections.impl.utility.Iterate.partition; -import static com.gs.collections.impl.utility.MapIterate.forEachKeyValue; - import java.io.IOException; import java.util.Arrays; import java.util.Collection; -import java.util.HashSet; import java.util.Map; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import com.gs.collections.api.RichIterable; +import com.gs.collections.api.block.function.Function; +import com.gs.collections.api.block.predicate.Predicate; +import com.gs.collections.api.list.ImmutableList; +import com.gs.collections.api.list.MutableList; +import com.gs.collections.api.multimap.list.ImmutableListMultimap; +import com.gs.collections.api.multimap.set.MutableSetMultimap; +import com.gs.collections.api.partition.PartitionIterable; +import com.gs.collections.api.set.MutableSet; +import com.gs.collections.api.tuple.Pair; +import com.gs.collections.impl.block.factory.Functions; +import com.gs.collections.impl.block.function.checked.CheckedFunction; +import com.gs.collections.impl.factory.Lists; +import com.gs.collections.impl.factory.Sets; +import com.gs.collections.impl.list.mutable.FastList; +import com.gs.collections.impl.map.mutable.UnifiedMap; +import com.gs.collections.impl.utility.ArrayIterate; +import com.gs.collections.impl.utility.Iterate; +import kafka.common.ErrorMapping; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.context.SmartLifecycle; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.kafka.core.BrokerAddress; import org.springframework.integration.kafka.core.ConnectionFactory; import org.springframework.integration.kafka.core.ConsumerException; @@ -47,26 +61,6 @@ import org.springframework.integration.kafka.core.Partition; import org.springframework.integration.kafka.core.Result; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; - -import com.gs.collections.api.RichIterable; -import com.gs.collections.api.block.function.Function; -import com.gs.collections.api.block.predicate.Predicate; -import com.gs.collections.api.block.procedure.Procedure; -import com.gs.collections.api.block.procedure.Procedure2; -import com.gs.collections.api.collection.MutableCollection; -import com.gs.collections.api.list.ImmutableList; -import com.gs.collections.api.list.MutableList; -import com.gs.collections.api.multimap.MutableMultimap; -import com.gs.collections.api.partition.PartitionIterable; -import com.gs.collections.impl.block.factory.Functions; -import com.gs.collections.impl.block.function.checked.CheckedFunction; -import com.gs.collections.impl.factory.Lists; -import com.gs.collections.impl.factory.Multimaps; -import com.gs.collections.impl.list.mutable.FastList; -import com.gs.collections.impl.utility.Iterate; - -import kafka.common.ErrorMapping; /** * @author Marius Bogoevici @@ -79,16 +73,12 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private static final Log log = LogFactory.getLog(KafkaMessageListenerContainer.class); - public static final Function, Partition> keyFunction = Functions.getKeyFunction(); - private final GetOffsetForPartitionFunction getOffset = new GetOffsetForPartitionFunction(); private final PartitionToLeaderFunction getLeader = new PartitionToLeaderFunction(); private final Function passThru = Functions.getPassThru(); - private final LaunchFetchTaskProcedure launchFetchTask = new LaunchFetchTaskProcedure(); - private final Object lifecycleMonitor = new Object(); private final KafkaTemplate kafkaTemplate; @@ -101,7 +91,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private Executor fetchTaskExecutor; - private Executor adminTaskExecutor = Executors.newSingleThreadExecutor(); + private Executor adminTaskExecutor; private Executor dispatcherTaskExecutor; @@ -125,7 +115,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private ConcurrentMessageListenerDispatcher messageDispatcher; - private final MutableMultimap partitionsByBrokerMap = Multimaps.mutable.set.with(); + private final ConcurrentMap fetchTasksByBroker = new ConcurrentHashMap<>(); private boolean autoCommitOnError = false; @@ -159,11 +149,10 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } public void setMessageListener(Object messageListener) { - Assert.isTrue - (messageListener instanceof MessageListener - || messageListener instanceof AcknowledgingMessageListener, - "Either a " + MessageListener.class.getName() + " or a " - + AcknowledgingMessageListener.class.getName() + " must be provided"); + Assert.isTrue( + messageListener instanceof MessageListener || messageListener instanceof AcknowledgingMessageListener, + "Either a " + MessageListener.class.getName() + " or a " + AcknowledgingMessageListener.class.getName() + + " must be provided"); this.messageListener = messageListener; } @@ -180,8 +169,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } /** - * The maximum number of concurrent {@link MessageListener}s running. Messages from within the same - * partition will be processed sequentially. + * The maximum number of concurrent {@link MessageListener}s running. Messages from + * within the same partition will be processed sequentially. * @param concurrency the concurrency maximum number */ public void setConcurrency(int concurrency) { @@ -189,7 +178,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } /** - * The timeout for waiting for each concurrent {@link MessageListener} to finish on stopping. + * The timeout for waiting for each concurrent {@link MessageListener} to finish on + * stopping. * @param stopTimeout timeout in milliseconds * @since 1.1 */ @@ -206,21 +196,20 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } /** - * The task executor for fetch operations + * The task executor for fetch operations. * @param fetchTaskExecutor the Executor for fetch operations */ public void setFetchTaskExecutor(Executor fetchTaskExecutor) { this.fetchTaskExecutor = fetchTaskExecutor; } - public Executor getAdminTaskExecutor() { return adminTaskExecutor; } /** - * The task executor for leader and offset updates - * @param adminTaskExecutor the task executor for leader and offset updates. + * The task executor for leader, offset, and partition reassignment updates. + * @param adminTaskExecutor the task executor for leader, offset and partition reassignment updates */ public void setAdminTaskExecutor(Executor adminTaskExecutor) { this.adminTaskExecutor = adminTaskExecutor; @@ -246,9 +235,9 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } /** - * The maximum number of messages that are buffered by each concurrent {@link MessageListener} runner. - * Increasing the value may increase throughput, but also increases the memory consumption. - * Must be a positive number and a power of 2. + * The maximum number of messages that are buffered by each concurrent + * {@link MessageListener} runner. Increasing the value may increase throughput, but + * also increases the memory consumption. Must be a positive number and a power of 2. * @param queueSize the queue size */ public void setQueueSize(int queueSize) { @@ -268,7 +257,6 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { * component will try to continue processing incoming messages. In the latter case, it is possible that * a successful message will commit an offset after a series of failures, so the component should rely on * the `errorHandler` to capture failures. - * * @param autoCommitOnError false if offsets should be committed only for successful messages * @since 1.3 */ @@ -326,12 +314,20 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { Arrays.asList(partitions), offsetManager, concurrency, queueSize, dispatcherTaskExecutor, autoCommitOnError); this.messageDispatcher.start(); - partitionsByBrokerMap.clear(); - partitionsByBrokerMap.putAll(partitionsAsList.groupBy(getLeader)); + fetchTasksByBroker.clear(); + ImmutableListMultimap partitionsByLeader = partitionsAsList + .groupBy(getLeader); if (fetchTaskExecutor == null) { - fetchTaskExecutor = Executors.newFixedThreadPool(partitionsByBrokerMap.keysView().size()); + fetchTaskExecutor = new SimpleAsyncTaskExecutor("kafka-fetch-"); + } + if (adminTaskExecutor == null) { + adminTaskExecutor = Executors.newSingleThreadExecutor(); + } + for (Pair> entry : partitionsByLeader.keyMultiValuePairsView()) { + FetchTask fetchTask = new FetchTask(entry.getOne(), entry.getTwo()); + fetchTaskExecutor.execute(fetchTask); + fetchTasksByBroker.put(entry.getOne(), fetchTask); } - partitionsByBrokerMap.forEachKey(launchFetchTask); } } } @@ -352,7 +348,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } private static Partition[] getPartitionsForTopics(final ConnectionFactory connectionFactory, String[] topics) { - MutableList partitionList = flatCollect(topics, new GetPartitionsForTopic(connectionFactory)); + MutableList partitionList = + ArrayIterate.flatCollect(topics, new GetPartitionsForTopic(connectionFactory)); return partitionList.toArray(new Partition[partitionList.size()]); } @@ -363,8 +360,21 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private final BrokerAddress brokerAddress; - public FetchTask(BrokerAddress brokerAddress) { + private final MutableSet listenedPartitions = Sets.mutable.of().asSynchronized(); + + private volatile boolean active; + + private final PartitionToFetchRequestFunction partitionToFetchRequestFunction = + new PartitionToFetchRequestFunction(); + + private final IsLeaderErrorPredicate isLeaderPredicate = new IsLeaderErrorPredicate(); + + private final IsOffsetOutOfRangePredicate offsetOutOfRangePredicate = new IsOffsetOutOfRangePredicate(); + + public FetchTask(BrokerAddress brokerAddress, RichIterable initialPartitions) { this.brokerAddress = brokerAddress; + this.active = true; + this.listenedPartitions.addAll(initialPartitions.toSet()); } @Override @@ -372,117 +382,104 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { return true; } + public boolean addListenedPartitionsIfActive(Iterable partitions) { + synchronized (listenedPartitions) { + if (active) { + listenedPartitions.addAllIterable(partitions); + } + return active; + } + } + @Override public void run() { - boolean wasInterrupted = false; - while (isRunning()) { - MutableCollection fetchPartitions; - synchronized (partitionsByBrokerMap) { - // retrieve the partitions for the current polling cycle - fetchPartitions = partitionsByBrokerMap.get(brokerAddress); - // do not proceed until there is something to read from - while (isRunning() && CollectionUtils.isEmpty(fetchPartitions)) { + try { + while (active && isRunning()) { + synchronized (listenedPartitions) { try { - // we only got here because there were no partitions to read from, - // so block until there is a change this prevents FetchTasks - // from busy waiting while leaders or offsets are being refreshed - // TODO: ideally we should use separate monitors for each task - partitionsByBrokerMap.wait(); - // see if the changes affect us - fetchPartitions = partitionsByBrokerMap.get(brokerAddress); - } - catch (InterruptedException e) { - wasInterrupted = true; - } - } - } - // we've just exited a potentially blocking operation. Is the component still running? - if (isRunning()) { - Set partitionsWithRemainingData; - boolean hasErrors; - do { - partitionsWithRemainingData = new HashSet(); - hasErrors = false; - try { - MutableCollection fetchRequests = - fetchPartitions.collect(new PartitionToFetchRequestFunction()); - Result result = kafkaTemplate.receive(fetchRequests); - // process successful messages first - Iterable batches = result.getResults().values(); - for (KafkaMessageBatch batch : batches) { - if (!batch.getMessages().isEmpty()) { - long highestFetchedOffset = 0; - for (KafkaMessage kafkaMessage : batch.getMessages()) { - // fetch operations may return entire blocks of compressed messages, - // which may have lower offsets than the ones requested - // thus a batch may contain messages that have been processed already - if (kafkaMessage.getMetadata().getOffset() >= fetchOffsets.get(batch.getPartition())) { - messageDispatcher.dispatch(kafkaMessage); - } - highestFetchedOffset = - Math.max(highestFetchedOffset, kafkaMessage.getMetadata().getNextOffset()); - } - fetchOffsets.replace(batch.getPartition(), highestFetchedOffset); - // if there are still messages on server, we can go on and retrieve more - if (highestFetchedOffset < batch.getHighWatermark()) { - partitionsWithRemainingData.add(batch.getPartition()); - } + if (!listenedPartitions.isEmpty()) { + Result result = fetchAvailableData(); + handleSuccessful(result); + if (result.getErrors().size() > 0) { + handleErrors(result); } } - // handle errors - if (result.getErrors().size() > 0) { - hasErrors = true; - - // find partitions with leader errors and - PartitionIterable> partitionByLeaderErrors = - partition(result.getErrors().entrySet(), new IsLeaderErrorPredicate()); - RichIterable partitionsWithLeaderErrors = - partitionByLeaderErrors.getSelected().collect(keyFunction); - resetLeaders(partitionsWithLeaderErrors); - - PartitionIterable> partitionsWithOffsetsOutOfRange = - partitionByLeaderErrors.getRejected() - .partition(new IsOffsetOutOfRangePredicate()); - resetOffsets(partitionsWithOffsetsOutOfRange.getSelected() - .collect(keyFunction) - .toSet()); - // it's not a leader issue - stopFetchingFromPartitions(partitionsWithOffsetsOutOfRange.getRejected() - .collect(keyFunction)); + else { + active = false; } } catch (ConsumerException e) { - resetLeaders(fetchPartitions.toImmutable()); + active = false; + // the connection is broken, terminate the task + kafkaTemplate.getConnectionFactory().disconnect(brokerAddress); + resetLeaders(listenedPartitions.toImmutable()); } - } while (!hasErrors && isRunning() && !partitionsWithRemainingData.isEmpty()); + } } } - if (wasInterrupted) { - Thread.currentThread().interrupt(); + finally { + active = false; + synchronized (fetchTasksByBroker) { + if (fetchTasksByBroker.get(brokerAddress) == this) { + fetchTasksByBroker.remove(brokerAddress); + } + } } } + private Result fetchAvailableData() { + return kafkaTemplate.receive(listenedPartitions.collect(partitionToFetchRequestFunction)); + } + + private void handleSuccessful(Result result) { + Iterable batches = result.getResults().values(); + for (KafkaMessageBatch batch : batches) { + if (!batch.getMessages().isEmpty()) { + long highestFetchedOffset = 0; + for (KafkaMessage kafkaMessage : batch.getMessages()) { + // fetch operations may return entire blocks of compressed messages, + // which may have lower offsets than the ones requested + // thus a batch may contain messages that have been processed already + if (kafkaMessage.getMetadata().getOffset() >= fetchOffsets.get(batch.getPartition())) { + messageDispatcher.dispatch(kafkaMessage); + } + highestFetchedOffset = Math.max(highestFetchedOffset, kafkaMessage.getMetadata().getNextOffset()); + } + fetchOffsets.replace(batch.getPartition(), highestFetchedOffset); + } + } + } + + private void handleErrors(Result result) { + Map errors = result.getErrors(); + PartitionIterable> splitByLeaderError = + Iterate.partition(errors.entrySet(), isLeaderPredicate); + RichIterable partitionsWithLeaderErrors = splitByLeaderError.getSelected() + .collect(Functions.getKeyFunction()); + resetLeaders(partitionsWithLeaderErrors); + PartitionIterable> splitByOffsetError = + splitByLeaderError.getRejected().partition(offsetOutOfRangePredicate); + RichIterable partitionsWithWrongOffsets = + splitByOffsetError.getSelected().collect(Functions.getKeyFunction()); + resetOffsets(partitionsWithWrongOffsets.toSet()); + // it's not a leader issue, remove everything else + RichIterable remainingPartitionsWithErrors + = splitByOffsetError.getRejected().collect(Functions.getKeyFunction()); + listenedPartitions.removeAllIterable(remainingPartitionsWithErrors); + } private void resetLeaders(final Iterable partitionsToReset) { - stopFetchingFromPartitions(partitionsToReset); + listenedPartitions.removeAllIterable(partitionsToReset); adminTaskExecutor.execute(new UpdateLeadersTask(partitionsToReset)); } - private void resetOffsets(final Collection partitionsToResetOffsets) { - stopFetchingFromPartitions(partitionsToResetOffsets); + listenedPartitions.removeAllIterable(partitionsToResetOffsets); adminTaskExecutor.execute(new UpdateOffsetsTask(partitionsToResetOffsets)); } - private void stopFetchingFromPartitions(Iterable partitions) { - synchronized (partitionsByBrokerMap) { - for (Partition partition : partitions) { - partitionsByBrokerMap.remove(brokerAddress, partition); - } - } - } - private class UpdateLeadersTask implements SchedulingAwareRunnable { + private final Iterable partitionsToReset; public UpdateLeadersTask(Iterable partitionsToReset) { @@ -503,10 +500,23 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { FastList partitionsAsList = FastList.newList(partitionsToReset); FastList topics = partitionsAsList.collect(new PartitionToTopicFunction()).distinct(); kafkaTemplate.getConnectionFactory().refreshMetadata(topics); - Map leaders = kafkaTemplate.getConnectionFactory().getLeaders(partitionsToReset); - synchronized (partitionsByBrokerMap) { - forEachKeyValue(leaders, new AddPartitionToBrokerProcedure()); - partitionsByBrokerMap.notifyAll(); + + MutableSetMultimap partitionsByBroker = UnifiedMap + .newMap(kafkaTemplate.getConnectionFactory().getLeaders(partitionsToReset)).flip(); + for (Pair> pair : partitionsByBroker + .keyMultiValuePairsView()) { + synchronized (fetchTasksByBroker) { + boolean addedSuccessfully = false; + FetchTask fetchTask = fetchTasksByBroker.get(pair.getOne()); + if (fetchTask != null) { + addedSuccessfully = fetchTask.addListenedPartitionsIfActive(pair.getTwo()); + } + if (!addedSuccessfully) { + fetchTask = new FetchTask(pair.getOne(), pair.getTwo()); + fetchTaskExecutor.execute(fetchTask); + fetchTasksByBroker.put(pair.getOne(), fetchTask); + } + } } fetchCompleted = true; } @@ -517,7 +527,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } catch (InterruptedException e1) { Thread.currentThread().interrupt(); - log.error("Interrupted after refresh leaders failure for: " + Iterate.makeString(partitionsToReset,",")); + log.error("Interrupted after refresh leaders failure for: " + Iterate + .makeString(partitionsToReset, ",")); fetchCompleted = true; } } @@ -541,12 +552,17 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { for (Partition partition : partitionsToResetOffsets) { fetchOffsets.replace(partition, offsetManager.getOffset(partition)); } - synchronized (partitionsByBrokerMap) { - for (Partition partitionsToResetOffset : partitionsToResetOffsets) { - partitionsByBrokerMap.put(brokerAddress, partitionsToResetOffset); + synchronized (fetchTasksByBroker) { + boolean addedSuccessfully = false; + FetchTask fetchTask = fetchTasksByBroker.get(brokerAddress); + if (fetchTask != null) { + addedSuccessfully = fetchTask.addListenedPartitionsIfActive(partitionsToResetOffsets); + } + if (!addedSuccessfully) { + fetchTask = new FetchTask(brokerAddress, Sets.immutable.ofAll(partitionsToResetOffsets)); + fetchTaskExecutor.execute(fetchTask); + fetchTasksByBroker.put(brokerAddress, fetchTask); } - // notify any waiting task that the partition allocation has changed - partitionsByBrokerMap.notifyAll(); } } @@ -557,8 +573,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { @Override public boolean accept(Map.Entry each) { - return each.getValue() == ErrorMapping.NotLeaderForPartitionCode() - || each.getValue() == ErrorMapping.UnknownTopicOrPartitionCode(); + return each.getValue() == ErrorMapping.NotLeaderForPartitionCode() || each.getValue() == ErrorMapping + .UnknownTopicOrPartitionCode(); } } @@ -600,16 +616,6 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } - @SuppressWarnings("serial") - private class LaunchFetchTaskProcedure implements Procedure { - - @Override - public void value(BrokerAddress brokerAddress) { - fetchTaskExecutor.execute(new FetchTask(brokerAddress)); - } - - } - @SuppressWarnings("serial") private class PartitionToFetchRequestFunction implements Function { @@ -646,14 +652,4 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } - @SuppressWarnings("serial") - private class AddPartitionToBrokerProcedure implements Procedure2 { - - @Override - public void value(Partition partition, BrokerAddress newBrokerAddress) { - partitionsByBrokerMap.put(newBrokerAddress, partition); - } - - } - } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java index c60d48fced..906189616f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java @@ -23,8 +23,8 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Properties; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import com.gs.collections.api.RichIterable; import com.gs.collections.api.block.function.Function2; @@ -38,11 +38,13 @@ import kafka.utils.TestUtils; import org.I0Itec.zkclient.ZkClient; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.junit.After; +import org.junit.Assert; import scala.collection.JavaConversions; import scala.collection.Map; import scala.collection.immutable.List$; @@ -66,6 +68,8 @@ public abstract class AbstractBrokerTests { public static final String TEST_TOPIC = "test-topic"; + private static final long SEND_TIMEOUT = 10000; + public abstract KafkaRule getKafkaRule(); @After @@ -136,6 +140,16 @@ public abstract class AbstractBrokerTests { return new Sender<>(producer); } + public Sender createMessageSender(String compression, int brokerIndex) { + Properties producerConfig = new Properties(); + producerConfig.setProperty("bootstrap.servers", getKafkaRule().getBrokerAddresses()[brokerIndex].toString()); + producerConfig.setProperty("compression.type", compression); + KafkaProducer producer = new KafkaProducer<>(producerConfig, + new EncoderAdaptingSerializer<>(new StringEncoder()), + new EncoderAdaptingSerializer<>(new StringEncoder())); + return new Sender<>(producer); + } + public ConnectionFactory getKafkaBrokerConnectionFactory() throws Exception { DefaultConnectionFactory connectionFactory = new DefaultConnectionFactory(getKafkaConfiguration()); connectionFactory.afterPropertiesSet(); @@ -175,21 +189,30 @@ public abstract class AbstractBrokerTests { } public void send(Collection> records) { - Future lastFuture = null; + final CountDownLatch sendLatch = new CountDownLatch(records.size()); + final ArrayList exceptions = new ArrayList<>(); for (ProducerRecord record : records) { - lastFuture = producer.send(record); + producer.send(record, new Callback() { + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + sendLatch.countDown(); + if (exception != null) { + exceptions.add(exception); + } + } + }); } - // only block if there is at least one message to be sent - if (lastFuture != null) { - try { - // block until the last message has been sent, so we make this deterministic - lastFuture.get(); - } catch (InterruptedException e) { - // not being able to confirm that all messages have been sent, fail the test - throw new RuntimeException(e); - } catch (ExecutionException e) { - throw new RuntimeException(e); + try { + sendLatch.await(SEND_TIMEOUT, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + if (exceptions.size() > 0) { + for (Exception exception : exceptions) { + log.error("Error while sending messages:", exception); } + Assert.fail("Cannot send messages"); } } } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/NewBrokerRecoveryTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/NewBrokerRecoveryTests.java new file mode 100644 index 0000000000..9046078e5f --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/NewBrokerRecoveryTests.java @@ -0,0 +1,136 @@ +/* + * Copyright 2015 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.hamcrest.collection.IsCollectionWithSize.hasSize; +import static org.hamcrest.core.IsEqual.equalTo; +import static org.junit.Assert.assertThat; +import static org.springframework.integration.kafka.util.MessageUtils.decodeKey; +import static org.springframework.integration.kafka.util.MessageUtils.decodePayload; + +import java.util.ArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import com.gs.collections.api.multimap.list.MutableListMultimap; +import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; +import kafka.serializer.StringDecoder; +import kafka.utils.VerifiableProperties; +import org.junit.Rule; +import org.junit.Test; + +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.KafkaMessage; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.rule.KafkaEmbedded; + +/** + * @author Marius Bogoevici + */ + +public class NewBrokerRecoveryTests extends AbstractMessageListenerContainerTests { + + @Rule + public KafkaEmbedded kafkaEmbeddedBrokerRule = new KafkaEmbedded(2); + + @Override + public KafkaEmbedded getKafkaRule() { + return kafkaEmbeddedBrokerRule; + } + + @Test + public void testCompleteShutdown() throws Exception { + int partitionCount = 1; + + createTopic(TEST_TOPIC, partitionCount, 2, 2); + + // stop one Kafka instance - the partition is now underreplicated and only one + kafkaEmbeddedBrokerRule.bounce(0); + + // connect to the live broker only + ConnectionFactory connectionFactory = + new DefaultConnectionFactory(new BrokerAddressListConfiguration(kafkaEmbeddedBrokerRule.getBrokerAddress(1))); + ArrayList readPartitions = new ArrayList<>(); + readPartitions.add(new Partition(TEST_TOPIC, 0)); + final KafkaMessageListenerContainer kafkaMessageListenerContainer = + new KafkaMessageListenerContainer(connectionFactory, + readPartitions.toArray(new Partition[readPartitions.size()])); + kafkaMessageListenerContainer.setMaxFetch(100); + kafkaMessageListenerContainer.setConcurrency(1); + + final int expectedMessageCount = 200; + createMessageSender("none", 1).send(createMessagesInRange(0, 49, TEST_TOPIC, partitionCount)); + + final MutableListMultimap receivedData = + new SynchronizedPutFastListMultimap(); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + kafkaMessageListenerContainer.setMessageListener(new MessageListener() { + + @Override + public void onMessage(KafkaMessage message) { + StringDecoder decoder = new StringDecoder(new VerifiableProperties()); + receivedData.put(message.getMetadata().getPartition().getId(), + new KeyedMessageWithOffset(decodeKey(message, decoder), decodePayload(message, decoder), + message.getMetadata().getOffset(), Thread.currentThread().getName(), + message.getMetadata().getPartition().getId())); + latch.countDown(); + } + + }); + + + kafkaMessageListenerContainer.start(); + + // now start sending messages again + createMessageSender("none", 1).send(createMessagesInRange(50, 99, TEST_TOPIC, partitionCount)); + + // start the other Kafka instance + kafkaEmbeddedBrokerRule.restart(0); + // sleep to let the brokers sync up + kafkaEmbeddedBrokerRule.waitUntilSynced(TEST_TOPIC, 0); + // bounce the other server + kafkaEmbeddedBrokerRule.bounce(1); + + // now start sending messages again + createMessageSender("none", 0).send(createMessagesInRange(100, 149, TEST_TOPIC, partitionCount)); + + // stop the other Kafka instance + kafkaEmbeddedBrokerRule.restart(1); + // sleep to let the brokers sync up + kafkaEmbeddedBrokerRule.waitUntilSynced(TEST_TOPIC, 1); + + // bounce the other server + kafkaEmbeddedBrokerRule.bounce(0); + + // now start sending messages again + createMessageSender("none", 1).send(createMessagesInRange(149, 199, TEST_TOPIC, partitionCount)); + + + latch.await(50, TimeUnit.SECONDS); + kafkaMessageListenerContainer.stop(); + + assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount)); + assertThat(latch.getCount(), equalTo(0L)); + + validateMessageReceipt(receivedData, 1, partitionCount, expectedMessageCount, 1); + + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java index 1eb5b87e47..a262aa1186 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java @@ -100,6 +100,7 @@ public class SingleBrokerRecoveryTests extends AbstractMessageListenerContainerT // restart Kafka kafkaEmbeddedBrokerRule.restart(0); + kafkaEmbeddedBrokerRule.waitUntilSynced(TEST_TOPIC,0); // now start sending messages again createMessageSender("none").send(createMessages(90, TEST_TOPIC,partitionCount)); @@ -109,7 +110,6 @@ public class SingleBrokerRecoveryTests extends AbstractMessageListenerContainerT assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount)); assertThat(latch.getCount(), equalTo(0L)); - System.out.println("All messages received ... checking "); validateMessageReceipt(receivedData, 1, partitionCount, expectedMessageCount, 1); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java index e4618355c5..6fd52c2b08 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java @@ -17,14 +17,20 @@ package org.springframework.integration.kafka.rule; -import static scala.collection.JavaConversions.asScalaBuffer; - import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Properties; +import com.gs.collections.api.block.function.Function; +import com.gs.collections.impl.list.mutable.FastList; +import com.gs.collections.impl.utility.ListIterate; +import kafka.admin.AdminUtils$; +import kafka.api.PartitionMetadata; +import kafka.api.TopicMetadata; +import kafka.cluster.Broker; import kafka.server.KafkaConfig; import kafka.server.KafkaServer; import kafka.server.NotRunning; @@ -34,10 +40,14 @@ import kafka.utils.TestZKUtils; import kafka.utils.Utils; import kafka.utils.ZKStringSerializer$; import kafka.zk.EmbeddedZookeeper; - import org.I0Itec.zkclient.ZkClient; import org.I0Itec.zkclient.exception.ZkInterruptedException; +import org.apache.kafka.common.protocol.Errors; import org.junit.rules.ExternalResource; +import scala.collection.JavaConversions; +import scala.collection.Map; +import scala.collection.Set; + import org.springframework.integration.kafka.core.BrokerAddress; import org.springframework.retry.RetryCallback; import org.springframework.retry.RetryContext; @@ -45,12 +55,6 @@ import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; -import scala.collection.JavaConversions; - -import com.gs.collections.api.block.function.Function; -import com.gs.collections.impl.list.mutable.FastList; -import com.gs.collections.impl.utility.ListIterate; - /** * @author Marius Bogoevici * @author Artem Bilan @@ -58,6 +62,8 @@ import com.gs.collections.impl.utility.ListIterate; @SuppressWarnings("serial") public class KafkaEmbedded extends ExternalResource implements KafkaRule { + public static final long METADATA_PROPAGATION_TIMEOUT = 10000L; + private int count; private boolean controlledShutdown; @@ -154,6 +160,11 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { return zookeeper.connectString(); } + public BrokerAddress getBrokerAddress(int i) { + KafkaServer kafkaServer = this.kafkaServers.get(i); + return new BrokerAddress(kafkaServer.config().hostName(),kafkaServer.config().port()); + } + @Override public BrokerAddress[] getBrokerAddresses() { return ListIterate.collect(this.kafkaServers, @@ -184,8 +195,36 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { public void bounce(int index, boolean waitForPropagation) { kafkaServers.get(index).shutdown(); if (waitForPropagation) { - TestUtils.waitUntilMetadataIsPropagated(asScalaBuffer(kafkaServers), "test-topic", 0, 5000L); + long initialTime = System.currentTimeMillis(); + boolean canExit = false; + do { + try { + Thread.sleep(100); + } + catch (InterruptedException e) { + break; + } + canExit = true; + Map topicProperties = AdminUtils$.MODULE$.fetchAllTopicConfigs(getZkClient()); + Set topicMetadatas = + AdminUtils$.MODULE$.fetchTopicMetadataFromZk(topicProperties.keySet(), getZkClient()); + for (TopicMetadata topicMetadata : JavaConversions.asJavaCollection(topicMetadatas)) { + if (Errors.forCode(topicMetadata.errorCode()).exception() == null) { + for (PartitionMetadata partitionMetadata : + JavaConversions.asJavaCollection(topicMetadata.partitionsMetadata())) { + Collection inSyncReplicas = JavaConversions.asJavaCollection(partitionMetadata.isr()); + for (Broker broker : inSyncReplicas) { + if (broker.id() == index) { + canExit = false; + } + } + } + } + } + } + while (!canExit && (System.currentTimeMillis() - initialTime < METADATA_PROPAGATION_TIMEOUT)); } + } public void bounce(int index) { @@ -212,13 +251,44 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { retryTemplate.execute(new RetryCallback() { @Override public Void doWithRetry(RetryContext context) throws Exception { - System.out.println("Retrying restart"); kafkaServers.get(index).startup(); return null; } }); } + public void waitUntilSynced(String topic, int brokerId) { + long initialTime = System.currentTimeMillis(); + boolean canExit = false; + do { + try { + Thread.sleep(100); + } + catch (InterruptedException e) { + break; + } + canExit = true; + TopicMetadata topicMetadata = AdminUtils$.MODULE$.fetchTopicMetadataFromZk(topic, getZkClient()); + if (Errors.forCode(topicMetadata.errorCode()).exception() == null) { + for (PartitionMetadata partitionMetadata : + JavaConversions.asJavaCollection(topicMetadata.partitionsMetadata())) { + Collection isr = JavaConversions.asJavaCollection(partitionMetadata.isr()); + boolean containsIndex = false; + for (Broker broker : isr) { + if (broker.id() == brokerId) { + containsIndex = true; + } + } + if (!containsIndex) { + canExit = false; + } + + } + } + } + while (!canExit && (System.currentTimeMillis() - initialTime < METADATA_PROPAGATION_TIMEOUT)); + } + @Override public String getBrokersAsString() { return FastList.newList(Arrays.asList(getBrokerAddresses())) diff --git a/spring-integration-kafka/src/test/resources/log4j.properties b/spring-integration-kafka/src/test/resources/log4j.properties index f7ce85a065..f5f5f3efe7 100644 --- a/spring-integration-kafka/src/test/resources/log4j.properties +++ b/spring-integration-kafka/src/test/resources/log4j.properties @@ -7,3 +7,4 @@ log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m log4j.category.org.springframework.integration=WARN log4j.category.org.springframework.integration.kafka=INFO log4j.category.org.apache.kafka.common.network.Selector=ERROR +log4j.category.kafka.server.ReplicaFetcherThread=ERROR