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

INTEXT-165 Recovery when cluster inaccessible

Browse Source 
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
This commit is contained in:
Marius Bogoevici
2015-04-24 18:47:40 +03:00
committed by Artem Bilan
parent df652634a5
commit fbe2975ed4
4 changed files with 212 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -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<Partition> partitionsAsList = FastList.newList(partitionsToReset);
FastList<String> topics = partitionsAsList.collect(new PartitionToTopicFunction()).distinct();
kafkaTemplate.getConnectionFactory().refreshMetadata(topics);
Map<Partition, BrokerAddress> 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<Partition> partitionsAsList = FastList.newList(partitionsToReset);
FastList<String> topics = partitionsAsList.collect(new PartitionToTopicFunction()).distinct();
kafkaTemplate.getConnectionFactory().refreshMetadata(topics);
Map<Partition, BrokerAddress> 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;
}
}
}
}
}

4
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java
View File

@@ -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<KeyedMessage<String, String>> createMessages(int count, String topic) {

116
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java Normal file
View File

@@ -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<Partition> readPartitions = new ArrayList<Partition>();
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<Integer, KeyedMessageWithOffset> receivedData =
new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
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);
}
}

51
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java
View File

@@ -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.<Class<? extends Throwable>,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<Void, Exception>() {
@Override
public Void doWithRetry(RetryContext context) throws Exception {
System.out.println("Retrying restart");
kafkaServers.get(index).startup();
return null;
}
});
}
@Override