From fbe2975ed400042ae4c76f4fd1db00eb75da7ef9 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Fri, 24 Apr 2015 18:47:40 +0300 Subject: [PATCH] INTEXT-165 Recovery when cluster inaccessible JIRA: https://jira.spring.io/browse/INTEXT-165 - retry fetching metadata until either successful or container stopped; - do not stop the fetch task on connection errors, and handle it as a case when all partitions have moved; - test broker stopped case + test enhancements for supporting broker restart; Unused imports Removed sysout calls from framework code Reduced socket timeout, to improve timing on failover tests --- .../KafkaMessageListenerContainer.java | 77 +++++++----- .../kafka/listener/AbstractBrokerTests.java | 4 +- .../listener/SingleBrokerRecoveryTests.java | 116 ++++++++++++++++++ .../integration/kafka/rule/KafkaEmbedded.java | 51 +++++++- 4 files changed, 212 insertions(+), 36 deletions(-) create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java 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 91be26ecd8..2c180617ac 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 @@ -31,6 +31,23 @@ 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.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; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -48,29 +65,13 @@ import org.springframework.integration.kafka.core.Result; 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 kafka.common.ErrorMapping; - /** * @author Marius Bogoevici */ public class KafkaMessageListenerContainer implements SmartLifecycle { + public static final int DEFAULT_WAIT_FOR_LEADER_REFRESH_RETRY = 5000; + private static final int DEFAULT_STOP_TIMEOUT = 1000; private static final Log log = LogFactory.getLog(KafkaMessageListenerContainer.class); @@ -406,14 +407,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } } catch (ConsumerException e) { - // this is a broker error, and we cannot recover from it. - // Reset leaders and stop fetching data from this broker altogether - log.error(e); resetLeaders(fetchPartitions.toImmutable()); - if (wasInterrupted) { - Thread.currentThread().interrupt(); - } - return; } } while (!hasErrors && isRunning() && !partitionsWithRemainingData.isEmpty()); } @@ -452,13 +446,32 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { @Override public void run() { - 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(); + // fetch can complete successfully or unsuccessfully + boolean fetchCompleted = false; + while (!fetchCompleted && isRunning()) { + try { + 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(); + } + fetchCompleted = true; + } + catch (Exception e) { + if (isRunning()) { + try { + Thread.sleep(DEFAULT_WAIT_FOR_LEADER_REFRESH_RETRY); + } + catch (InterruptedException e1) { + Thread.currentThread().interrupt(); + log.error("Interrupted after refresh leaders failure for: " + Iterate.makeString(partitionsToReset,",")); + fetchCompleted = true; + } + } + } } } 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 fe0782e817..894a4170d5 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 @@ -105,7 +105,9 @@ public abstract class AbstractBrokerTests { public Configuration getKafkaConfiguration() { - return new BrokerAddressListConfiguration(getKafkaRule().getBrokerAddresses()); + BrokerAddressListConfiguration configuration = new BrokerAddressListConfiguration(getKafkaRule().getBrokerAddresses()); + configuration.setSocketTimeout(500); + return configuration; } public static scala.collection.Seq> createMessages(int count, String topic) { 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 new file mode 100644 index 0000000000..0e9b9b7ee7 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java @@ -0,0 +1,116 @@ +/* + * 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.ConnectionFactory; +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 SingleBrokerRecoveryTests extends AbstractMessageListenerContainerTests { + + @Rule + public KafkaEmbedded kafkaEmbeddedBrokerRule = new KafkaEmbedded(1); + + @Override + public KafkaEmbedded getKafkaRule() { + return kafkaEmbeddedBrokerRule; + } + + @Test + public void testCompleteShutdown() throws Exception { + createTopic(TEST_TOPIC, 1, 1, 1); + + ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory(); + 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 = 100; + + createStringProducer(0).send(createMessages(10, TEST_TOPIC)); + + 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(); + + // stop Kafka + kafkaEmbeddedBrokerRule.bounce(0, false); + + // sleep one second to let things settle + Thread.sleep(1000); + + // restart Kafka + kafkaEmbeddedBrokerRule.restart(0); + + // now start sending messages again + createStringProducer(0).send(createMessages(90, TEST_TOPIC)); + + latch.await(50, TimeUnit.SECONDS); + kafkaMessageListenerContainer.stop(); + + assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount)); + assertThat(latch.getCount(), equalTo(0L)); + System.out.println("All messages received ... checking "); + + validateMessageReceipt(receivedData, 1, 1, 100, expectedMessageCount, readPartitions, 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 72ffa30132..7a414589eb 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 @@ -21,6 +21,7 @@ import static scala.collection.JavaConversions.asScalaBuffer; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Properties; @@ -45,6 +46,13 @@ import org.junit.rules.ExternalResource; import scala.collection.JavaConversions; import org.springframework.integration.kafka.core.BrokerAddress; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryPolicy; +import org.springframework.retry.backoff.BackOffPolicy; +import org.springframework.retry.backoff.ExponentialBackOffPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; /** * @author Marius Bogoevici @@ -78,7 +86,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { @Override protected void before() throws Throwable { - zookeeper = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect()); + startZookeeper(); int zkConnectionTimeout = 6000; int zkSessionTimeout = 6000; zookeeperClient = new ZkClient(TestZKUtils.zookeeperConnect(), zkSessionTimeout, zkConnectionTimeout, @@ -166,9 +174,46 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { } } - public void bounce(int index) { + public void startZookeeper() { + zookeeper = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect()); + } + + public void bounce(int index, boolean waitForPropagation) { kafkaServers.get(index).shutdown(); - TestUtils.waitUntilMetadataIsPropagated(asScalaBuffer(kafkaServers), "test-topic", 0, 5000L); + if (waitForPropagation) { + TestUtils.waitUntilMetadataIsPropagated(asScalaBuffer(kafkaServers), "test-topic", 0, 5000L); + } + } + + public void bounce(int index) { + bounce(index, true); + } + + public void restart(final int index) throws Exception { + + // retry restarting repeatedly, first attempts may fail + + SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(10, + Collections.,Boolean>singletonMap(Exception.class, true)); + + ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy(); + backOffPolicy.setInitialInterval(100); + backOffPolicy.setMaxInterval(1000); + backOffPolicy.setMultiplier(2); + + RetryTemplate retryTemplate = new RetryTemplate(); + retryTemplate.setRetryPolicy(retryPolicy); + retryTemplate.setBackOffPolicy(backOffPolicy); + + + retryTemplate.execute(new RetryCallback() { + @Override + public Void doWithRetry(RetryContext context) throws Exception { + System.out.println("Retrying restart"); + kafkaServers.get(index).startup(); + return null; + } + }); } @Override