From 1896ef8879461702ea4326263060f91343b6513c Mon Sep 17 00:00:00 2001 From: Yasuhiro Matsuda Date: Thu, 6 Aug 2015 11:22:26 -0700 Subject: [PATCH] implement StateStore methods in SlidingWindow --- .../kafka/stream/internals/StreamGroup.java | 3 +- .../kafka/stream/topology/SlidingWindow.java | 139 +++++++++++++-- .../topology/internals/WindowSupport.java | 159 ++++++++++++++++++ 3 files changed, 282 insertions(+), 19 deletions(-) create mode 100644 stream/src/main/java/org/apache/kafka/stream/topology/internals/WindowSupport.java diff --git a/stream/src/main/java/org/apache/kafka/stream/internals/StreamGroup.java b/stream/src/main/java/org/apache/kafka/stream/internals/StreamGroup.java index 7138668ec5127..5f7b52c0824e6 100644 --- a/stream/src/main/java/org/apache/kafka/stream/internals/StreamGroup.java +++ b/stream/src/main/java/org/apache/kafka/stream/internals/StreamGroup.java @@ -263,8 +263,9 @@ public int buffered() { } public void close() { - chooser.close(); + for (RecordQueue queue : stash.values()) queue.stream.close(); stash.clear(); + chooser.close(); } protected RecordQueue createRecordQueue(TopicPartition partition, KStreamSource stream) { diff --git a/stream/src/main/java/org/apache/kafka/stream/topology/SlidingWindow.java b/stream/src/main/java/org/apache/kafka/stream/topology/SlidingWindow.java index feb9a5d3cdd06..081a1c3b2702e 100644 --- a/stream/src/main/java/org/apache/kafka/stream/topology/SlidingWindow.java +++ b/stream/src/main/java/org/apache/kafka/stream/topology/SlidingWindow.java @@ -17,30 +17,66 @@ package org.apache.kafka.stream.topology; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.stream.KStreamContext; +import org.apache.kafka.stream.RecordCollector; +import org.apache.kafka.stream.RestoreFunc; +import org.apache.kafka.stream.topology.internals.WindowSupport; import org.apache.kafka.stream.util.FilteredIterator; import org.apache.kafka.stream.util.Stamped; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedList; +import java.util.Map; -public class SlidingWindow implements Window { +public class SlidingWindow extends WindowSupport implements Window { + private final Object lock = new Object(); + private final Serializer keySerializer; + private final Serializer valueSerializer; + private final Deserializer keyDeserializer; + private final Deserializer valueDeserializer; + private KStreamContext context; + private int slotNum; private String name; private final long duration; private final int maxCount; private LinkedList list = new LinkedList(); - private HashMap>> map = new HashMap>>(); - - public SlidingWindow(String name, long duration, int maxCount) { + private HashMap> map = new HashMap<>(); + + public SlidingWindow( + String name, + long duration, + int maxCount, + Serializer keySerializer, + Serializer valueSerializer, + Deserializer keyDeseriaizer, + Deserializer valueDeserializer) { this.name = name; this.duration = duration; this.maxCount = maxCount; + this.keySerializer = keySerializer; + this.valueSerializer = valueSerializer; + this.keyDeserializer = keyDeseriaizer; + this.valueDeserializer = valueDeserializer; } @Override public void init(KStreamContext context) { + this.context = context; + RestoreFuncImpl restoreFunc = new RestoreFuncImpl(); + context.register(this, restoreFunc); + + for (ValueList valueList : map.values()) { + valueList.clearDirtyValues(); + } + this.slotNum = restoreFunc.slotNum; } @Override @@ -62,14 +98,14 @@ public Iterator find(K key, final long timestamp) { * finds items in the window between startTime and endTime (both inclusive) */ private Iterator find(K key, final long startTime, final long endTime) { - final LinkedList> values = map.get(key); + final ValueList values = map.get(key); if (values == null) { return null; } else { - return new FilteredIterator>(values.iterator()) { + return new FilteredIterator>(values.iterator()) { @Override - protected V filter(Stamped item) { + protected V filter(Value item) { if (startTime <= item.timestamp && item.timestamp <= endTime) return item.value; else @@ -81,16 +117,19 @@ protected V filter(Stamped item) { @Override public void put(K key, V value, long timestamp) { - list.offerLast(key); + synchronized (lock) { + slotNum++; - LinkedList> values = map.get(key); - if (values == null) { - values = new LinkedList>(); - map.put(key, values); - } + list.offerLast(key); - values.offerLast(new Stamped(value, timestamp)); + ValueList values = map.get(key); + if (values == null) { + values = new ValueList<>(); + map.put(key, values); + } + values.add(slotNum, value, timestamp); + } evictExcess(); evictExpired(timestamp - duration); } @@ -99,7 +138,7 @@ private void evictExcess() { while (list.size() > maxCount) { K oldestKey = list.pollFirst(); - LinkedList> values = map.get(oldestKey); + ValueList values = map.get(oldestKey); values.removeFirst(); if (values.isEmpty()) map.remove(oldestKey); @@ -110,8 +149,8 @@ private void evictExpired(long cutoffTime) { while (true) { K oldestKey = list.peekFirst(); - LinkedList> values = map.get(oldestKey); - Stamped oldestValue = values.peekFirst(); + ValueList values = map.get(oldestKey); + Stamped oldestValue = values.first(); if (oldestValue.timestamp < cutoffTime) { list.pollFirst(); @@ -131,7 +170,38 @@ public String name() { @Override public void flush() { - // TODO + IntegerSerializer intSerializer = new IntegerSerializer(); + ByteArraySerializer byteArraySerializer = new ByteArraySerializer(); + + RecordCollector collector = context.recordCollector(); + + for (Map.Entry> entry : map.entrySet()) { + ValueList values = entry.getValue(); + if (values.hasDirtyValues()) { + K key = entry.getKey(); + + byte[] keyBytes = keySerializer.serialize(name, key); + + Iterator> iterator = values.dirtyValueIterator(); + while (iterator.hasNext()) { + Value dirtyValue = iterator.next(); + byte[] slot = intSerializer.serialize("", dirtyValue.slotNum); + byte[] valBytes = valueSerializer.serialize(name, dirtyValue.value); + + byte[] combined = new byte[8 + 4 + keyBytes.length + 4 + valBytes.length]; + + int offset = 0; + offset += putLong(combined, offset, dirtyValue.timestamp); + offset += puts(combined, offset, keyBytes); + offset += puts(combined, offset, valBytes); + + if (offset != combined.length) throw new IllegalStateException("serialized length does not match"); + + collector.send(new ProducerRecord<>(name, context.id(), slot, combined), byteArraySerializer, byteArraySerializer); + } + values.clearDirtyValues(); + } + } } @Override @@ -144,4 +214,37 @@ public boolean persistent() { // TODO: should not be persistent, right? return false; } + + private class RestoreFuncImpl implements RestoreFunc { + + final IntegerDeserializer intDeserializer; + int slotNum = 0; + + RestoreFuncImpl() { + intDeserializer = new IntegerDeserializer(); + } + + @Override + public void apply(byte[] slot, byte[] bytes) { + + slotNum = intDeserializer.deserialize("", slot); + + int offset = 0; + // timestamp + long timestamp = getLong(bytes, offset); + offset += 8; + // key + int length = getInt(bytes, offset); + offset += 4; + K key = deserialize(bytes, offset, length, name, keyDeserializer); + offset += length; + // value + length = getInt(bytes, offset); + offset += 4; + V value = deserialize(bytes, offset, length, name, valueDeserializer); + + put(key, value, timestamp); + } + } + } diff --git a/stream/src/main/java/org/apache/kafka/stream/topology/internals/WindowSupport.java b/stream/src/main/java/org/apache/kafka/stream/topology/internals/WindowSupport.java new file mode 100644 index 0000000000000..5ce2c7b75e363 --- /dev/null +++ b/stream/src/main/java/org/apache/kafka/stream/topology/internals/WindowSupport.java @@ -0,0 +1,159 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.kafka.stream.topology.internals; + +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.stream.util.Stamped; + +import java.util.Iterator; + +public class WindowSupport { + + public static class ValueList { + Value head = null; + Value tail = null; + Value dirty = null; + + public void add(int slotNum, V value, long timestamp) { + Value v = new Value<>(slotNum, value, timestamp); + if (tail != null) { + tail.next = v; + } else { + head = v; + } + tail = v; + if (dirty == null) dirty = v; + } + + public Value first() { + return head; + } + + public void removeFirst() { + if (head != null) { + if (head == tail) tail = null; + head = head.next; + } + } + + public boolean isEmpty() { + return head == null; + } + + public boolean hasDirtyValues() { + return dirty != null; + } + + public void clearDirtyValues() { + dirty = null; + } + + public Iterator> iterator() { + return new ValueListIterator(head); + } + + public Iterator> dirtyValueIterator() { + return new ValueListIterator(dirty); + } + + } + + private static class ValueListIterator implements Iterator> { + + Value ptr; + + ValueListIterator(Value start) { + ptr = start; + } + + @Override + public boolean hasNext() { + return ptr != null; + } + + @Override + public Value next() { + Value value = ptr; + if (value != null) ptr = value.next; + return value; + } + + @Override + public void remove() { + throw new UnsupportedOperationException(); + } + + } + + public static class Value extends Stamped { + public final int slotNum; + private Value next = null; + + Value(int slotNum, V value, long timestamp) { + super(value, timestamp); + this.slotNum = slotNum; + } + } + + + public static long getLong(byte[] bytes, int offset) { + long value = 0; + for (int i = 0; i < 8; i++) { + value = (value << 8) | bytes[offset + i]; + } + return value; + } + + public static int getInt(byte[] bytes, int offset) { + int value = 0; + for (int i = 0; i < 4; i++) { + value = (value << 8) | bytes[offset + i]; + } + return value; + } + + public static int putLong(byte[] bytes, int offset, long value) { + for (int i = 7; i >= 0; i--) { + bytes[offset + i] = (byte) (value & 0xFF); + value = value >> 8; + } + return 8; + } + + public static int putInt(byte[] bytes, int offset, int value) { + for (int i = 3; i >= 0; i--) { + bytes[offset + i] = (byte) (value & 0xFF); + value = value >> 8; + } + return 4; + } + + public static int puts(byte[] bytes, int offset, byte[] value) { + offset += putInt(bytes, offset, value.length); + System.arraycopy(bytes, offset, value, 0, value.length); + return 4 + value.length; + } + + + public static T deserialize(byte[] bytes, int offset, int length, String topic, Deserializer deserializer) { + byte[] buf = new byte[length]; + System.arraycopy(bytes, offset, buf, 0, length); + return deserializer.deserialize(topic, buf); + } + +}