INTEXT-220: Add WindowingOffsetManager
JIRA: https://jira.spring.io/browse/INTEXT-220 Add `WindowingOffsetManagerTests` Rework `WindowingOffsetManager` to be "resettable" on `flush()`. * `flush()` `WindowingOffsetManager` on each `getOffset()`, `resetOffsets()` and `deleteOffset()`. Exactly for the `RingBufferProcessor` processor completion to allow to have the access to "real" offset with those operations over `delegate`. * Since Reactor's `Stream` isn't restartable create an new instance recursively from the `onComplete()`
This commit is contained in:
Artem Bilan
@@ -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.
|
||||
* <p>
|
||||
* 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<Long, Long, Long> maxFunction = new BiFunction<Long, Long, Long>() {
|
||||
|
||||
@Override
|
||||
public Long apply(Long aLong, Long bLong) {
|
||||
return Math.max(aLong, bLong);
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
private static final Function<PartitionAndOffset, Long> offsetFunction
|
||||
= new Function<PartitionAndOffset, Long>() {
|
||||
|
||||
@Override
|
||||
public Long apply(PartitionAndOffset partitionAndOffset) {
|
||||
return partitionAndOffset.getOffset();
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
private static final ComputeMaximumOffsetByPartitionFunction findHighestOffsetInPartitionGroup
|
||||
= new ComputeMaximumOffsetByPartitionFunction();
|
||||
|
||||
private static final Function<PartitionAndOffset, Partition> getPartitionFunction
|
||||
= new Function<PartitionAndOffset, Partition>() {
|
||||
|
||||
@Override
|
||||
public Partition apply(PartitionAndOffset partitionAndOffset) {
|
||||
return partitionAndOffset.getPartition();
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
private static final FindHighestOffsetsByPartitionFunction findHighestOffsetsByPartition
|
||||
= new FindHighestOffsetsByPartitionFunction();
|
||||
|
||||
private final Consumer<PartitionAndOffset> delegateUpdateOffset = new Consumer<PartitionAndOffset>() {
|
||||
|
||||
@Override
|
||||
public void accept(PartitionAndOffset partitionAndOffset) {
|
||||
delegate.updateOffset(partitionAndOffset.getPartition(), partitionAndOffset.getOffset());
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
private final Consumer<Void> offsetComplete = new Consumer<Void>() {
|
||||
|
||||
@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<PartitionAndOffset> 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> 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<GroupedStream<Partition, PartitionAndOffset>, Stream<PartitionAndOffset>> {
|
||||
|
||||
@Override
|
||||
public Stream<PartitionAndOffset> apply(final GroupedStream<Partition, PartitionAndOffset> group) {
|
||||
return group
|
||||
.map(offsetFunction)
|
||||
.reduce(maxFunction)
|
||||
.map(new Function<Long, PartitionAndOffset>() {
|
||||
|
||||
@Override
|
||||
public PartitionAndOffset apply(Long offset) {
|
||||
return new PartitionAndOffset(group.key(), offset);
|
||||
}
|
||||
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static class FindHighestOffsetsByPartitionFunction
|
||||
implements Function<Stream<PartitionAndOffset>, Stream<PartitionAndOffset>> {
|
||||
|
||||
@Override
|
||||
public Stream<PartitionAndOffset> apply(Stream<PartitionAndOffset> windowBuffer) {
|
||||
return windowBuffer
|
||||
.groupBy(getPartitionFunction)
|
||||
.flatMap(findHighestOffsetInPartitionGroup);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<Partition, Long> 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;
|
||||
}
|
||||
}
|
||||
@@ -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<Integer, Integer> partitionDistribution =
|
||||
createPartitionDistribution(partitionCount, brokers, replication);
|
||||
ensureTopicCreated(zkClient, topicName, partitionCount, new Properties(),
|
||||
toKafkaPartitionMap(partitionDistribution));
|
||||
}
|
||||
|
||||
|
||||
public MutableMultimap<Integer, Integer> createPartitionDistribution(int partitionCount, int brokers,
|
||||
int replication) {
|
||||
MutableMultimap<Integer, Integer> 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<Integer, Integer> partitions) {
|
||||
java.util.Map<Object, Seq<Object>> m = partitions.toMap()
|
||||
.collect(new Function2<Integer, RichIterable<Integer>, Pair<Object, Seq<Object>>>() {
|
||||
|
||||
@Override
|
||||
public Pair<Object, Seq<Object>> value(Integer argument1, RichIterable<Integer> argument2) {
|
||||
return Tuples.pair((Object) argument1,
|
||||
List$.MODULE$.fromArray(argument2.toArray(new Object[0])).toSeq());
|
||||
}
|
||||
|
||||
});
|
||||
return Map$.MODULE$.apply(JavaConversions.asScalaMap(m).toSeq());
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user