From d8a1115f1f03bfd7eb5a7fc485fb7e027ac1c71b Mon Sep 17 00:00:00 2001 From: Chris Lemper Date: Thu, 27 Aug 2015 13:33:21 -0600 Subject: [PATCH] Kafka native offset implementation Fixes #79 (https://github.com/spring-projects/spring-integration-kafka/pull/79) Adds an implementation for offset management based on Kafka's own offset management support Update copyright and use fill name in author tag. Refactoring the native offset manager Remove unused imports Polishing --- .../kafka/core/AbstractConfiguration.java | 15 +- .../integration/kafka/core/Configuration.java | 8 + .../kafka/core/DefaultConnection.java | 2 +- .../kafka/core/KafkaConsumerDefaults.java | 4 +- .../listener/KafkaNativeOffsetManager.java | 231 ++++++++++++++++++ .../outbound/KafkaProducerMessageHandler.java | 3 +- .../offset/KafkaNativeOffsetManagerTests.java | 48 ++++ ...java => KafkaTopicOffsetManagerTests.java} | 2 +- .../integration/kafka/rule/KafkaEmbedded.java | 1 + 9 files changed, 308 insertions(+), 6 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaNativeOffsetManager.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaNativeOffsetManagerTests.java rename spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/{KafkaOffsetManagerTests.java => KafkaTopicOffsetManagerTests.java} (95%) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java index 1d85fd94c8..047b5730d2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java @@ -47,8 +47,9 @@ public abstract class AbstractConfiguration implements InitializingBean, Configu private int socketTimeout = KafkaConsumerDefaults.SOCKET_TIMEOUT_INT; - private int fetchMetadataTimeout = KafkaConsumerDefaults.FETCH_METADATA_TIMEOUT; + private int backoff = KafkaConsumerDefaults.BACKOFF_INCREMENT_INT; + private int fetchMetadataTimeout = KafkaConsumerDefaults.FETCH_METADATA_TIMEOUT; /** * The minimum amount of data that a server fetch operation will wait for before returning, @@ -126,6 +127,18 @@ public abstract class AbstractConfiguration implements InitializingBean, Configu this.socketTimeout = socketTimeout; } + @Override + public int getBackOff() { + return backoff; + } + + /** + * The retry backoff time for this client + * @param backoff the retry backoff + */ + public void setBackoff(int backoff) { + this.backoff = backoff; + } /** * The timeout on fetching metadata (e.g. partition leaders) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java index e4de8b96fb..da4342ec49 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java @@ -66,6 +66,13 @@ public interface Configuration { */ int getSocketTimeout(); + /** + * The retry backoff for this client + * @return the retry backoff + * @since 1.3 + */ + int getBackOff(); + /** * The timeout on fetching metadata (e.g. partition leaders) * @return the fetch metadata timeout @@ -90,4 +97,5 @@ public interface Configuration { */ String getDefaultTopic(); + } 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 a1f49bb109..c11a15aeec 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 @@ -317,7 +317,7 @@ public class DefaultConnection implements Connection { @Override public Pair value(Partition partition, Long offset) { return Tuples.pair(new TopicAndPartition(partition.getTopic(), partition.getId()), - new OffsetAndMetadata(offset, OffsetAndMetadata.NoMetadata(), ErrorMapping.NoError())); + new OffsetAndMetadata(offset, OffsetAndMetadata.NoMetadata(), OffsetAndMetadata.InvalidTime())); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/KafkaConsumerDefaults.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/KafkaConsumerDefaults.java index 2b332b7ca4..6953064cbb 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/KafkaConsumerDefaults.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/KafkaConsumerDefaults.java @@ -42,7 +42,9 @@ public abstract class KafkaConsumerDefaults { public static final String FETCH_SIZE = Integer.toString(FETCH_SIZE_INT); - public static final String BACKOFF_INCREMENT = "1000"; + public static final int BACKOFF_INCREMENT_INT = 1000; + + public static final String BACKOFF_INCREMENT = Integer.toString(BACKOFF_INCREMENT_INT); public static final String QUEUED_CHUNKS_MAX = "100"; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaNativeOffsetManager.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaNativeOffsetManager.java new file mode 100644 index 0000000000..8725d6d173 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaNativeOffsetManager.java @@ -0,0 +1,231 @@ +/* + * 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 java.io.IOException; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.I0Itec.zkclient.ZkClient; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.kafka.core.BrokerAddress; +import org.springframework.integration.kafka.core.Configuration; +import org.springframework.integration.kafka.core.Connection; +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.core.ConsumerException; +import org.springframework.integration.kafka.core.DefaultConnectionFactory; +import org.springframework.integration.kafka.core.KafkaConsumerDefaults; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.core.PartitionNotFoundException; +import org.springframework.integration.kafka.core.Result; +import org.springframework.integration.kafka.support.ZookeeperConnect; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.backoff.ExponentialBackOffPolicy; +import org.springframework.retry.listener.RetryListenerSupport; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +import com.gs.collections.impl.factory.Maps; + +import kafka.client.ClientUtils$; +import kafka.common.ErrorMapping; +import kafka.common.OffsetAndMetadata; +import kafka.network.BlockingChannel; +import kafka.utils.ZKStringSerializer$; + +/** + * Implementation of an {@link OffsetManager} that uses kafka native 'topic' offset storage. + * + * @author Chris Lemper + * @author Marius Bogoevici + * @since 1.3 + */ +public class KafkaNativeOffsetManager extends AbstractOffsetManager implements InitializingBean { + + private static final String PARTITION_ATTRIBUTE = "partition"; + + private final Map offsetManagerBrokerAddressCache = new ConcurrentHashMap<>(); + + private final ZkClient zkClient; + + private RetryTemplate retryTemplate; + + /** + * @param connectionFactory a Kafka connection factory + * @param zookeeperConnect the zookeeper connection information + */ + public KafkaNativeOffsetManager(ConnectionFactory connectionFactory, ZookeeperConnect zookeeperConnect) { + this(connectionFactory, zookeeperConnect, Collections.emptyMap()); + } + + /** + * @param connectionFactory a Kafka connection factory + * @param zookeeperConnect zookeeper connection for retrieving + * @param initialOffsets a map of partitions to initial offsets + */ + public KafkaNativeOffsetManager(ConnectionFactory connectionFactory, ZookeeperConnect zookeeperConnect, + Map initialOffsets) { + super(connectionFactory, initialOffsets); + Assert.notNull(zookeeperConnect, "'zookeeperConnect' must not be null."); + this.zkClient = new ZkClient(zookeeperConnect.getZkConnect(), + Integer.parseInt(zookeeperConnect.getZkSessionTimeout()), + Integer.parseInt(zookeeperConnect.getZkConnectionTimeout()), + ZKStringSerializer$.MODULE$); + } + + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.retryTemplate = retryTemplate; + } + + @Override + public void afterPropertiesSet() throws Exception { + if (this.retryTemplate == null) { + this.retryTemplate = new RetryTemplate(); + this.retryTemplate.registerListener(new ResetOffsetManagerBrokerAddressRetryListener()); + final SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(); + retryPolicy.setMaxAttempts(5); + this.retryTemplate.setRetryPolicy(retryPolicy); + final ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy(); + backOffPolicy.setInitialInterval(125L); + backOffPolicy.setMaxInterval(5000L); + backOffPolicy.setMultiplier(2); + this.retryTemplate.setBackOffPolicy(backOffPolicy); + } + } + + @Override + protected Long doGetOffset(final Partition partition) { + final Long offset = this.retryTemplate.execute(new RetryCallback() { + + @Override + public Long doWithRetry(RetryContext context) throws RuntimeException { + context.setAttribute(PARTITION_ATTRIBUTE, partition); + Result result = getOffsetManagerConnection(partition).fetchStoredOffsetsForConsumer( + getConsumerId(), partition); + checkResultForErrors(result, partition); + return result.getResult(partition); + } + + @Override + public String toString() { + return String.format("fetchStoredOffsetsForConsumer(%s, %s)", getConsumerId(), partition); + } + + }); + + if (offset != null && offset < 0) { + return null; + } + + return offset; + } + + @Override + protected void doUpdateOffset(final Partition partition, final long offset) { + this.retryTemplate.execute(new RetryCallback() { + + @Override + public Void doWithRetry(RetryContext context) throws RuntimeException { + context.setAttribute(PARTITION_ATTRIBUTE, partition); + Result result = getOffsetManagerConnection(partition).commitOffsetsForConsumer( + getConsumerId(), Maps.immutable.of(partition, offset).castToMap()); + checkResultForErrors(result, partition); + return null; + } + + @Override + public String toString() { + return String.format("commitOffsetsForConsumer(%s, %s, %s)", getConsumerId(), partition, offset); + } + + }); + } + + @Override + protected void doRemoveOffset(Partition partition) { + doUpdateOffset(partition, OffsetAndMetadata.InvalidOffset()); + } + + @Override + public void close() throws IOException { + this.zkClient.close(); + } + + @Override + public void flush() throws IOException { + // this function left intentionally blank + } + + private BrokerAddress getOffsetManagerBrokerAddress(final Partition partition) { + BrokerAddress brokerAddress = this.offsetManagerBrokerAddressCache.get(partition); + if (brokerAddress == null) { + int socketTimeoutMs = KafkaConsumerDefaults.SOCKET_TIMEOUT_INT; + int retryBackOffMs = KafkaConsumerDefaults.BACKOFF_INCREMENT_INT; + if (this.connectionFactory instanceof DefaultConnectionFactory) { + Configuration configuration = ((DefaultConnectionFactory) connectionFactory).getConfiguration(); + socketTimeoutMs = configuration.getSocketTimeout(); + retryBackOffMs = configuration.getBackOff(); + } + final BlockingChannel channel = ClientUtils$.MODULE$.channelToOffsetManager( + getConsumerId(), zkClient, socketTimeoutMs, retryBackOffMs); + brokerAddress = new BrokerAddress(channel.host(), channel.port()); + if (log.isDebugEnabled()) { + log.debug(String.format("Offset manager for [%s] is at [%s].", partition, + brokerAddress)); + } + this.offsetManagerBrokerAddressCache.put(partition, brokerAddress); + channel.disconnect(); + } + return brokerAddress; + } + + private Connection getOffsetManagerConnection(final Partition partition) { + return this.connectionFactory.connect(getOffsetManagerBrokerAddress(partition)); + } + + private void checkResultForErrors(final Result result, final Partition partition) { + if (result.getErrors().containsKey(partition)) { + final short errorCode = result.getError(partition); + if (errorCode == ErrorMapping.UnknownTopicOrPartitionCode()) { + throw new PartitionNotFoundException(partition); + } + else if (errorCode != ErrorMapping.NoError()) { + throw new ConsumerException(ErrorMapping.exceptionFor(errorCode)); + } + } + } + + private class ResetOffsetManagerBrokerAddressRetryListener extends RetryListenerSupport { + + @Override + @SuppressWarnings("unchecked") + public void onError(RetryContext context, RetryCallback callback, Throwable t) { + if (log.isWarnEnabled()) { + log.warn("Retrying kafka operation [" + callback + "] due to [" + t + "]", t); + } + Partition partition = (Partition) context.getAttribute(PARTITION_ATTRIBUTE); + if (partition != null) { + offsetManagerBrokerAddressCache.remove(partition); + } + } + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index e9f9b0e6c9..27810a9c76 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -19,12 +19,10 @@ package org.springframework.integration.kafka.outbound; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.integration.expression.ExpressionUtils; -import org.springframework.integration.expression.IntegrationEvaluationContextAware; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.kafka.support.KafkaHeaders; import org.springframework.integration.kafka.support.KafkaProducerContext; import org.springframework.messaging.Message; -import org.springframework.util.Assert; /** * @author Soby Chacko @@ -58,6 +56,7 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { } /** + * @param partitionExpression an expression that returns a partition id * @deprecated as of 1.3, {@link #setPartitionIdExpression(Expression)} should be used instead */ @Deprecated diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaNativeOffsetManagerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaNativeOffsetManagerTests.java new file mode 100644 index 0000000000..f37ba4222a --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaNativeOffsetManagerTests.java @@ -0,0 +1,48 @@ +/* + * 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.offset; + +import java.util.Map; + +import org.springframework.integration.kafka.core.DefaultConnectionFactory; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.core.ZookeeperConfiguration; +import org.springframework.integration.kafka.listener.KafkaNativeOffsetManager; +import org.springframework.integration.kafka.listener.OffsetManager; +import org.springframework.integration.kafka.support.ZookeeperConnect; + +/** + * @author Chris Lemper + * @since 1.3 + */ +public class KafkaNativeOffsetManagerTests extends AbstractOffsetManagerTests { + + @Override + protected OffsetManager createOffsetManager(long referenceTimestamp, String consumerId, + Map initialOffsets) throws Exception { + ZookeeperConnect zookeeperConnect = new ZookeeperConnect(kafkaRule.getZookeeperConnectionString()); + KafkaNativeOffsetManager kafkaNativeOffsetManager = + new KafkaNativeOffsetManager(new DefaultConnectionFactory(new ZookeeperConfiguration(zookeeperConnect)), + zookeeperConnect, + initialOffsets); + kafkaNativeOffsetManager.setConsumerId(consumerId); + kafkaNativeOffsetManager.afterPropertiesSet(); + kafkaNativeOffsetManager.setReferenceTimestamp(referenceTimestamp); + return kafkaNativeOffsetManager; + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaTopicOffsetManagerTests.java similarity index 95% rename from spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java rename to spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaTopicOffsetManagerTests.java index cd0e6f94ff..17ed99a4ae 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaTopicOffsetManagerTests.java @@ -26,7 +26,7 @@ import org.springframework.integration.kafka.support.ZookeeperConnect; /** * @author Marius Bogoevici */ -public class KafkaOffsetManagerTests extends AbstractOffsetManagerTests { +public class KafkaTopicOffsetManagerTests extends AbstractOffsetManagerTests { @Override protected OffsetManager createOffsetManager(long referenceTimestamp, String consumerId, 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 1be43e8683..e4618355c5 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 @@ -93,6 +93,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { Properties brokerConfigProperties = TestUtils.createBrokerConfig(i, kafkaPorts.get(i),controlledShutdown); brokerConfigProperties.setProperty("replica.socket.timeout.ms","1000"); brokerConfigProperties.setProperty("controller.socket.timeout.ms","1000"); + brokerConfigProperties.setProperty("offsets.topic.replication.factor","1"); KafkaServer server = TestUtils.createServer(new KafkaConfig(brokerConfigProperties), SystemTime$.MODULE$); kafkaServers.add(server); }