diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/WindowingOffsetManager.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/WindowingOffsetManager.java new file mode 100644 index 0000000000..5cf33a98ff --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/WindowingOffsetManager.java @@ -0,0 +1,310 @@ +/* + * Copyright 2016 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.Collection; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.util.Assert; + +import reactor.Environment; +import reactor.core.processor.RingBufferProcessor; +import reactor.fn.BiFunction; +import reactor.fn.Consumer; +import reactor.fn.Function; +import reactor.rx.Stream; +import reactor.rx.Streams; +import reactor.rx.stream.GroupedStream; + +/** + * An {@link OffsetManager} that aggregates writes over a time or count window, using an underlying delegate to + * do the actual operations. Its purpose is to reduce the performance impact of writing operations + * wherever this is desirable. + *

+ * A time window or a number of writes can be specified, or both. + * Defaults to 10 seconds window with {@link Integer#MAX_VALUE} buffer. + * @author Marius Bogoevici + * @author Artem Bilan + * @since 1.3.1 + */ +public class WindowingOffsetManager implements OffsetManager, InitializingBean, DisposableBean { + + static { + Environment.initializeIfEmpty(); + } + + private static final BiFunction maxFunction = new BiFunction() { + + @Override + public Long apply(Long aLong, Long bLong) { + return Math.max(aLong, bLong); + } + + }; + + private static final Function offsetFunction + = new Function() { + + @Override + public Long apply(PartitionAndOffset partitionAndOffset) { + return partitionAndOffset.getOffset(); + } + + }; + + private static final ComputeMaximumOffsetByPartitionFunction findHighestOffsetInPartitionGroup + = new ComputeMaximumOffsetByPartitionFunction(); + + private static final Function getPartitionFunction + = new Function() { + + @Override + public Partition apply(PartitionAndOffset partitionAndOffset) { + return partitionAndOffset.getPartition(); + } + + }; + + private static final FindHighestOffsetsByPartitionFunction findHighestOffsetsByPartition + = new FindHighestOffsetsByPartitionFunction(); + + private final Consumer delegateUpdateOffset = new Consumer() { + + @Override + public void accept(PartitionAndOffset partitionAndOffset) { + delegate.updateOffset(partitionAndOffset.getPartition(), partitionAndOffset.getOffset()); + } + + }; + + private final Consumer offsetComplete = new Consumer() { + + @Override + public void accept(Void aVoid) { + createOffsetsStream(); + } + + }; + + private final ReadWriteLock offsetsLock = new ReentrantReadWriteLock(); + + private final OffsetManager delegate; + + private long timespan = 10 * 1000; + + private int count = Integer.MAX_VALUE; + + private int shutdownTimeout = 2000; + + private volatile RingBufferProcessor offsets; + + private volatile boolean closed; + + public WindowingOffsetManager(OffsetManager offsetManager) { + this.delegate = offsetManager; + } + + /** + * The timespan for aggregating write operations, before invoking the underlying {@link OffsetManager}. + * @param timespan duration in milliseconds + */ + public void setTimespan(long timespan) { + Assert.isTrue(timespan >= 0, "Timespan must be a positive value"); + this.timespan = timespan; + } + + /** + * How many writes should be aggregated, before invoking the underlying {@link OffsetManager}. Setting this value + * to 1 effectively disables windowing. + * @param count number of writes + */ + public void setCount(int count) { + Assert.isTrue(count >= 0, "Count must be a positive value"); + this.count = count; + } + + /** + * The timeout that {@link #close()} and {@link #destroy()} + * operations will wait for receiving a confirmation that + * the underlying writes have been processed. + * @param shutdownTimeout duration in milliseconds + */ + public void setShutdownTimeout(int shutdownTimeout) { + this.shutdownTimeout = shutdownTimeout; + } + + @Override + public void afterPropertiesSet() throws Exception { + if (this.count != 1) { + createOffsetsStream(); + } + } + + private void createOffsetsStream() { + if (!this.closed) { + this.offsetsLock.writeLock().lock(); + try { + this.offsets = RingBufferProcessor.share("spring-integration-kafka-offset", 1024); + } + finally { + this.offsetsLock.writeLock().unlock(); + } + Streams.wrap(this.offsets) + .window(this.count, timespan, TimeUnit.MILLISECONDS) + .flatMap(findHighestOffsetsByPartition) + .consume(this.delegateUpdateOffset, null, this.offsetComplete); + + } + } + + @Override + public void destroy() throws Exception { + flush(); + close(); + if (this.delegate instanceof DisposableBean) { + ((DisposableBean) this.delegate).destroy(); + } + } + + @Override + public void updateOffset(Partition partition, long offset) { + if (this.offsets != null) { + this.offsetsLock.readLock().lock(); + try { + this.offsets.onNext(new PartitionAndOffset(partition, offset)); + } + finally { + this.offsetsLock.readLock().unlock(); + } + } + else { + this.delegate.updateOffset(partition, offset); + } + } + + @Override + public long getOffset(Partition partition) { + doFlush(); + return this.delegate.getOffset(partition); + } + + @Override + public void deleteOffset(Partition partition) { + doFlush(); + this.delegate.deleteOffset(partition); + } + + @Override + public void resetOffsets(Collection partition) { + doFlush(); + this.delegate.resetOffsets(partition); + } + + @Override + public void close() throws IOException { + this.closed = true; + this.delegate.close(); + } + + @Override + public void flush() throws IOException { + if (this.offsets != null) { + this.offsets.awaitAndShutdown(this.shutdownTimeout, TimeUnit.MILLISECONDS); + } + this.delegate.flush(); + } + + private void doFlush() { + try { + flush(); + } + catch (IOException e) { + throw new IllegalStateException(e); + } + } + + + private static class PartitionAndOffset { + + private final Partition partition; + + private final Long offset; + + public PartitionAndOffset(Partition partition, Long offset) { + this.partition = partition; + this.offset = offset; + } + + public Partition getPartition() { + return partition; + } + + public Long getOffset() { + return offset; + } + + @Override + public String toString() { + return "PartitionAndOffset{" + + "partition=" + partition + + ", offset=" + offset + + '}'; + } + + } + + + private static class ComputeMaximumOffsetByPartitionFunction + implements Function, Stream> { + + @Override + public Stream apply(final GroupedStream group) { + return group + .map(offsetFunction) + .reduce(maxFunction) + .map(new Function() { + + @Override + public PartitionAndOffset apply(Long offset) { + return new PartitionAndOffset(group.key(), offset); + } + + }); + } + + } + + private static class FindHighestOffsetsByPartitionFunction + implements Function, Stream> { + + @Override + public Stream apply(Stream windowBuffer) { + return windowBuffer + .groupBy(getPartitionFunction) + .flatMap(findHighestOffsetInPartitionGroup); + } + + } + +} + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/WindowingOffsetManagerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/WindowingOffsetManagerTests.java new file mode 100644 index 0000000000..a7e25f86a7 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/WindowingOffsetManagerTests.java @@ -0,0 +1,54 @@ +/* + * Copyright 2016 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.listener.WindowingOffsetManager; +import org.springframework.integration.kafka.support.ZookeeperConnect; + +/** + * @author Artem Bilan + * @since 1.3.1 + */ +public class WindowingOffsetManagerTests 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); + + WindowingOffsetManager windowingOffsetManager = new WindowingOffsetManager(kafkaNativeOffsetManager); + windowingOffsetManager.setCount(2); + windowingOffsetManager.setTimespan(10); + windowingOffsetManager.afterPropertiesSet(); + return windowingOffsetManager; + } +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/performance/OffsetManagerPerformanceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/performance/OffsetManagerPerformanceTests.java new file mode 100644 index 0000000000..fb28c7a6c3 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/performance/OffsetManagerPerformanceTests.java @@ -0,0 +1,154 @@ +/* + * Copyright 2016 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.performance; + +import static org.hamcrest.CoreMatchers.is; +import static org.springframework.integration.kafka.util.TopicUtils.ensureTopicCreated; + +import java.util.Properties; + +import org.I0Itec.zkclient.ZkClient; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +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.KafkaTopicOffsetManager; +import org.springframework.integration.kafka.listener.OffsetManager; +import org.springframework.integration.kafka.listener.WindowingOffsetManager; +import org.springframework.integration.kafka.rule.KafkaEmbedded; +import org.springframework.integration.kafka.support.ZookeeperConnect; +import org.springframework.util.StopWatch; + +import com.gs.collections.api.RichIterable; +import com.gs.collections.api.block.function.Function2; +import com.gs.collections.api.multimap.Multimap; +import com.gs.collections.api.multimap.MutableMultimap; +import com.gs.collections.api.tuple.Pair; +import com.gs.collections.impl.factory.Multimaps; +import com.gs.collections.impl.tuple.Tuples; +import scala.collection.JavaConversions; +import scala.collection.Map; +import scala.collection.immutable.List$; +import scala.collection.immutable.Map$; +import scala.collection.immutable.Seq; + +/** + * @author Marius Bogoevici + * @author Artem Bilan + */ +public class OffsetManagerPerformanceTests { + + public static final int UPDATE_COUNT = 100000; + + @Rule + public KafkaEmbedded embedded = new KafkaEmbedded(1); + + private final StopWatch stopWatch = new StopWatch("OffsetManagers Performance"); + + @Test + public void testPerformance() throws Exception { + ZookeeperConnect zookeeperConnect = new ZookeeperConnect(embedded.getZookeeperConnectionString()); + + KafkaTopicOffsetManager kafkaTopicOffsetManager = + new KafkaTopicOffsetManager(zookeeperConnect, "zkTopic"); + + KafkaNativeOffsetManager kafkaNativeOffsetManager = new KafkaNativeOffsetManager( + new DefaultConnectionFactory(new ZookeeperConfiguration(zookeeperConnect)), zookeeperConnect); + + KafkaTopicOffsetManager topicOffsetManagerForAssertion = + new KafkaTopicOffsetManager(zookeeperConnect, "zkTopic"); + + KafkaNativeOffsetManager nativeOffsetManagerForAssertion = new KafkaNativeOffsetManager( + new DefaultConnectionFactory(new ZookeeperConfiguration(zookeeperConnect)), zookeeperConnect); + + doTest("Just KafkaTopicOffsetManager", kafkaTopicOffsetManager, topicOffsetManagerForAssertion, false); + doTest("Just KafkaNativeOffsetManager", kafkaNativeOffsetManager, nativeOffsetManagerForAssertion, false); + doTest("Window KafkaTopicOffsetManager", kafkaTopicOffsetManager, topicOffsetManagerForAssertion, true); + doTest("Window KafkaNativeOffsetManager", kafkaNativeOffsetManager, nativeOffsetManagerForAssertion, true); + + System.out.println(this.stopWatch.prettyPrint()); + } + + private void doTest(String description, OffsetManager offsetManager, + OffsetManager offsetManagerForAssertion, boolean window) throws Exception { + createTopic(this.embedded.getZkClient(), "sometopic", 1, 1, 1); + Partition partition = new Partition("sometopic", 0); + + ((InitializingBean) offsetManager).afterPropertiesSet(); + + offsetManager.updateOffset(partition, 0); + + if (window) { + WindowingOffsetManager windowingOffsetManager = new WindowingOffsetManager(offsetManager); + windowingOffsetManager.setCount(100); + windowingOffsetManager.afterPropertiesSet(); + offsetManager = windowingOffsetManager; + } + + this.stopWatch.start(description); + for (long i = 1; i < UPDATE_COUNT; i++) { + offsetManager.updateOffset(partition, i); + } + stopWatch.stop(); + ((DisposableBean) offsetManager).destroy(); + + ((InitializingBean) offsetManagerForAssertion).afterPropertiesSet(); + Assert.assertThat(offsetManagerForAssertion.getOffset(partition), is(99999L)); + } + + @SuppressWarnings("unchecked") + public void createTopic(ZkClient zkClient, String topicName, int partitionCount, int brokers, int replication) { + MutableMultimap partitionDistribution = + createPartitionDistribution(partitionCount, brokers, replication); + ensureTopicCreated(zkClient, topicName, partitionCount, new Properties(), + toKafkaPartitionMap(partitionDistribution)); + } + + + public MutableMultimap createPartitionDistribution(int partitionCount, int brokers, + int replication) { + MutableMultimap partitionDistribution = Multimaps.mutable.list.with(); + for (int i = 0; i < partitionCount; i++) { + for (int j = 0; j < replication; j++) { + partitionDistribution.put(i, (i + j) % brokers); + } + } + return partitionDistribution; + } + + @SuppressWarnings({"rawtypes", "serial", "deprecation", "unchecked"}) + private Map toKafkaPartitionMap(Multimap partitions) { + java.util.Map> m = partitions.toMap() + .collect(new Function2, Pair>>() { + + @Override + public Pair> value(Integer argument1, RichIterable argument2) { + return Tuples.pair((Object) argument1, + List$.MODULE$.fromArray(argument2.toArray(new Object[0])).toSeq()); + } + + }); + return Map$.MODULE$.apply(JavaConversions.asScalaMap(m).toSeq()); + } + +}