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

GH-70 Allow the client to discover new brokers

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-kafka/issues/70

* Refactor the message listener container to finish tasks on broker connection failure
and spawn new ones whenever a new broker is discovered;
* Add an explicit disconnect operation on the ConnectionFactory to release the connection
on failure;
* Other improvements and test refactorings;

Optimize the fetch logic

Add test for recognizing new brokers

Formatting

Formatting and refactoring

Test improvements, polishing
This commit is contained in:
Marius Bogoevici
2015-11-03 14:20:10 -05:00
committed by Artem Bilan
parent 4ff25fd894
commit d6a3b343aa
10 changed files with 487 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java
View File

@@ -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

21
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnection.java
View File

@@ -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<KafkaMessageBatch> resultBuilder = new ResultBuilder<KafkaMessageBatch>();
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) {

54
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java
View File

@@ -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<MetadataCache> metadataCacheHolder =
new AtomicReference<MetadataCache>(new MetadataCache(Collections.<TopicMetadata>emptySet()));
private final AtomicReference<MetadataCache> metadataCacheHolder = new AtomicReference<MetadataCache>(
new MetadataCache(Collections.<TopicMetadata>emptySet()));
private final ReadWriteLock lock = new ReentrantReadWriteLock();
@@ -169,20 +167,14 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac
public void refreshMetadata(Collection<String> 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<String>(topics)),
ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString),
this.configuration.getClientId(), this.configuration.getFetchMetadataTimeout(), 0));
PartitionIterable<TopicMetadata> 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<String>(topics)),
ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString),
this.configuration.getClientId(), this.configuration.getFetchMetadataTimeout(), 0));
PartitionIterable<TopicMetadata> 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)
*/

35
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java
View File

@@ -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<Partition> 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<Partition> 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<QueueingMessageListenerInvoker> delegateList = new ArrayList<QueueingMessageListenerInvoker>(consumers);
List<QueueingMessageListenerInvoker> 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));
}

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

@@ -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<Map.Entry<Partition, ?>, Partition> keyFunction = Functions.getKeyFunction();
private final GetOffsetForPartitionFunction getOffset = new GetOffsetForPartitionFunction();
private final PartitionToLeaderFunction getLeader = new PartitionToLeaderFunction();
private final Function<Partition, Partition> 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<BrokerAddress, Partition> partitionsByBrokerMap = Multimaps.mutable.set.with();
private final ConcurrentMap<BrokerAddress, FetchTask> 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<BrokerAddress, Partition> 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<BrokerAddress, RichIterable<Partition>> 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<Partition> partitionList = flatCollect(topics, new GetPartitionsForTopic(connectionFactory));
MutableList<Partition> 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<Partition> listenedPartitions = Sets.mutable.<Partition>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<Partition> 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<Partition> partitions) {
synchronized (listenedPartitions) {
if (active) {
listenedPartitions.addAllIterable(partitions);
}
return active;
}
}
@Override
public void run() {
boolean wasInterrupted = false;
while (isRunning()) {
MutableCollection<Partition> 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<Partition> partitionsWithRemainingData;
boolean hasErrors;
do {
partitionsWithRemainingData = new HashSet<Partition>();
hasErrors = false;
try {
MutableCollection<FetchRequest> fetchRequests =
fetchPartitions.collect(new PartitionToFetchRequestFunction());
Result<KafkaMessageBatch> result = kafkaTemplate.receive(fetchRequests);
// process successful messages first
Iterable<KafkaMessageBatch> 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<KafkaMessageBatch> 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<Map.Entry<Partition, Short>> partitionByLeaderErrors =
partition(result.getErrors().entrySet(), new IsLeaderErrorPredicate());
RichIterable<Partition> partitionsWithLeaderErrors =
partitionByLeaderErrors.getSelected().collect(keyFunction);
resetLeaders(partitionsWithLeaderErrors);
PartitionIterable<Map.Entry<Partition, Short>> 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<KafkaMessageBatch> fetchAvailableData() {
return kafkaTemplate.receive(listenedPartitions.collect(partitionToFetchRequestFunction));
}
private void handleSuccessful(Result<KafkaMessageBatch> result) {
Iterable<KafkaMessageBatch> 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<KafkaMessageBatch> result) {
Map<Partition, Short> errors = result.getErrors();
PartitionIterable<Map.Entry<Partition, Short>> splitByLeaderError =
Iterate.partition(errors.entrySet(), isLeaderPredicate);
RichIterable<Partition> partitionsWithLeaderErrors = splitByLeaderError.getSelected()
.collect(Functions.<Partition>getKeyFunction());
resetLeaders(partitionsWithLeaderErrors);
PartitionIterable<Map.Entry<Partition, Short>> splitByOffsetError =
splitByLeaderError.getRejected().partition(offsetOutOfRangePredicate);
RichIterable<Partition> partitionsWithWrongOffsets =
splitByOffsetError.getSelected().collect(Functions.<Partition>getKeyFunction());
resetOffsets(partitionsWithWrongOffsets.toSet());
// it's not a leader issue, remove everything else
RichIterable<Partition> remainingPartitionsWithErrors
= splitByOffsetError.getRejected().collect(Functions.<Partition>getKeyFunction());
listenedPartitions.removeAllIterable(remainingPartitionsWithErrors);
}
private void resetLeaders(final Iterable<Partition> partitionsToReset) {
stopFetchingFromPartitions(partitionsToReset);
listenedPartitions.removeAllIterable(partitionsToReset);
adminTaskExecutor.execute(new UpdateLeadersTask(partitionsToReset));
}
private void resetOffsets(final Collection<Partition> partitionsToResetOffsets) {
stopFetchingFromPartitions(partitionsToResetOffsets);
listenedPartitions.removeAllIterable(partitionsToResetOffsets);
adminTaskExecutor.execute(new UpdateOffsetsTask(partitionsToResetOffsets));
}
private void stopFetchingFromPartitions(Iterable<Partition> partitions) {
synchronized (partitionsByBrokerMap) {
for (Partition partition : partitions) {
partitionsByBrokerMap.remove(brokerAddress, partition);
}
}
}
private class UpdateLeadersTask implements SchedulingAwareRunnable {
private final Iterable<Partition> partitionsToReset;
public UpdateLeadersTask(Iterable<Partition> partitionsToReset) {
@@ -503,10 +500,23 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
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();
MutableSetMultimap<BrokerAddress, Partition> partitionsByBroker = UnifiedMap
.newMap(kafkaTemplate.getConnectionFactory().getLeaders(partitionsToReset)).flip();
for (Pair<BrokerAddress, RichIterable<Partition>> 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<Partition, Short> 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<BrokerAddress> {
@Override
public void value(BrokerAddress brokerAddress) {
fetchTaskExecutor.execute(new FetchTask(brokerAddress));
}
}
@SuppressWarnings("serial")
private class PartitionToFetchRequestFunction implements Function<Partition, FetchRequest> {
@@ -646,14 +652,4 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
}
@SuppressWarnings("serial")
private class AddPartitionToBrokerProcedure implements Procedure2<Partition, BrokerAddress> {
@Override
public void value(Partition partition, BrokerAddress newBrokerAddress) {
partitionsByBrokerMap.put(newBrokerAddress, partition);
}
}
}

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

@@ -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<String, String> createMessageSender(String compression, int brokerIndex) {
Properties producerConfig = new Properties();
producerConfig.setProperty("bootstrap.servers", getKafkaRule().getBrokerAddresses()[brokerIndex].toString());
producerConfig.setProperty("compression.type", compression);
KafkaProducer<String, String> 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<ProducerRecord<K,V>> records) {
Future<RecordMetadata> lastFuture = null;
final CountDownLatch sendLatch = new CountDownLatch(records.size());
final ArrayList<Exception> exceptions = new ArrayList<>();
for (ProducerRecord<K, V> 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");
}
}
}

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

@@ -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<Partition> 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<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();
// 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);
}
}

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerRecoveryTests.java
View File

@@ -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);

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

@@ -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<String, Properties> topicProperties = AdminUtils$.MODULE$.fetchAllTopicConfigs(getZkClient());
Set<TopicMetadata> 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<Broker> 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<Void, Exception>() {
@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<Broker> 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()))

1
spring-integration-kafka/src/test/resources/log4j.properties
View File

@@ -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