You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by wc...@apache.org on 2023/08/19 03:00:17 UTC
[kafka] branch trunk updated: KAFKA-14936: Add restore logic (3/N) (#14027)
This is an automated email from the ASF dual-hosted git repository.
wcarlson pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d0b7677c2c9 KAFKA-14936: Add restore logic (3/N) (#14027)
d0b7677c2c9 is described below
commit d0b7677c2c9a8265643a90175f86987521615849
Author: Walker Carlson <18...@users.noreply.github.com>
AuthorDate: Fri Aug 18 22:00:04 2023 -0500
KAFKA-14936: Add restore logic (3/N) (#14027)
Added restore logic for the buffer in grace period joins.
Reviewers: Matthias J. Sax <ma...@confluent.io>, Bill Bejeck <bb...@gmail.com>
---
.../streams/kstream/internals/KStreamImpl.java | 2 +-
.../RocksDBTimeOrderedKeyValueBuffer.java | 65 +++++-
.../RocksDBTimeOrderedKeyValueBytesStore.java | 29 ++-
.../JoinGracePeriodDurabilityIntegrationTest.java | 256 +++++++++++++++++++++
.../RocksDBTimeOrderedKeyValueBufferTest.java | 2 +-
.../RocksDBTimeOrderedKeyValueBytesStoreTest.java | 110 +++++++++
6 files changed, 459 insertions(+), 5 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
index 00d8b179ce2..3c127ae7f1e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
@@ -1271,7 +1271,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K, V> implements KStream<K
final String bufferStoreName = name + "-Buffer";
final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDBTimeOrderedKeyValueBytesStoreSupplier(bufferStoreName).get();
- buffer = Optional.of(new RocksDBTimeOrderedKeyValueBuffer<>(store, joined.gracePeriod(), name));
+ buffer = Optional.of(new RocksDBTimeOrderedKeyValueBuffer<>(store, joined.gracePeriod(), name, true));
}
final ProcessorSupplier<K, V, K, ? extends VR> processorSupplier = new KStreamKTableJoin<>(
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java
index fb62e3d3986..5edf6edf88a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
@@ -23,7 +25,10 @@ import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StateStoreContext;
import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorContextUtils;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.processor.internals.SerdeGetter;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.ValueAndTimestamp;
@@ -36,7 +41,8 @@ import java.util.function.Supplier;
import static java.util.Objects.requireNonNull;
public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<RocksDBTimeOrderedKeyValueBytesStore, Object, Object> implements TimeOrderedKeyValueBuffer<K, V, V> {
-
+ private static final BytesSerializer KEY_SERIALIZER = new BytesSerializer();
+ private static final ByteArraySerializer VALUE_SERIALIZER = new ByteArraySerializer();
private final long gracePeriod;
private long bufferSize;
private long minTimestamp;
@@ -45,10 +51,15 @@ public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<Ro
private Serde<V> valueSerde;
private final String topic;
private int seqnum;
+ private final boolean loggingEnabled;
+ private int partition;
+ private String changelogTopic;
+ private InternalProcessorContext context;
public RocksDBTimeOrderedKeyValueBuffer(final RocksDBTimeOrderedKeyValueBytesStore store,
final Duration gracePeriod,
- final String topic) {
+ final String topic,
+ final boolean loggingEnabled) {
super(store);
this.gracePeriod = gracePeriod.toMillis();
minTimestamp = Long.MAX_VALUE;
@@ -56,6 +67,7 @@ public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<Ro
bufferSize = 0;
seqnum = 0;
this.topic = topic;
+ this.loggingEnabled = loggingEnabled;
}
@SuppressWarnings("unchecked")
@@ -74,6 +86,11 @@ public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<Ro
@Override
public void init(final StateStoreContext context, final StateStore root) {
wrapped().init(context, wrapped());
+ this.context = ProcessorContextUtils.asInternalProcessorContext(context);
+ partition = context.taskId().partition();
+ if (loggingEnabled) {
+ changelogTopic = ProcessorContextUtils.changelogFor(context, name(), Boolean.TRUE);
+ }
}
@Override
@@ -103,6 +120,11 @@ public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<Ro
callback.accept(new Eviction<>(key, value, bufferValue.context()));
wrapped().remove(keyValue.key);
+
+ if (loggingEnabled) {
+ logTombstone(keyValue.key);
+ }
+
numRecords--;
bufferSize = bufferSize - computeRecordSize(keyValue.key, bufferValue);
}
@@ -137,6 +159,12 @@ public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<Ro
final byte[] valueBytes = valueSerde.serializer().serialize(topic, record.value());
final BufferValue buffered = new BufferValue(null, null, valueBytes, recordContext);
wrapped().put(serializedKey, buffered.serialize(0).array());
+
+ if (loggingEnabled) {
+ final BufferKey key = new BufferKey(0L, serializedKey);
+ logValue(serializedKey, key, buffered);
+ }
+
bufferSize += computeRecordSize(serializedKey, buffered);
numRecords++;
if (minTimestamp() > record.timestamp()) {
@@ -172,4 +200,37 @@ public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<Ro
private void maybeUpdateSeqnumForDups() {
seqnum = (seqnum + 1) & 0x7FFFFFFF;
}
+
+ private void logValue(final Bytes key, final BufferKey bufferKey, final BufferValue value) {
+ final int sizeOfBufferTime = Long.BYTES;
+ final ByteBuffer buffer = value.serialize(sizeOfBufferTime);
+ buffer.putLong(bufferKey.time());
+ final byte[] array = buffer.array();
+ ((RecordCollector.Supplier) context).recordCollector().send(
+ changelogTopic,
+ key,
+ array,
+ null,
+ partition,
+ null,
+ KEY_SERIALIZER,
+ VALUE_SERIALIZER,
+ null,
+ null);
+ }
+
+ private void logTombstone(final Bytes key) {
+ ((RecordCollector.Supplier) context).recordCollector().send(
+ changelogTopic,
+ key,
+ null,
+ null,
+ partition,
+ null,
+ KEY_SERIALIZER,
+ VALUE_SERIALIZER,
+ null,
+ null);
+ }
+
}
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
index 5906f8c8b9a..f2e562ae04d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
@@ -17,12 +17,16 @@
package org.apache.kafka.streams.state.internals;
import java.util.Collection;
+import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.errors.ProcessorStateException;
+import org.apache.kafka.streams.processor.internals.ChangelogRecordDeserializationHelper;
import org.apache.kafka.streams.state.internals.PrefixedWindowKeySchemas.TimeFirstWindowKeySchema;
+import org.rocksdb.RocksDBException;
import org.rocksdb.WriteBatch;
/**
@@ -42,7 +46,30 @@ public class RocksDBTimeOrderedKeyValueBytesStore extends AbstractRocksDBTimeOrd
@Override
Map<KeyValueSegment, WriteBatch> getWriteBatches(final Collection<ConsumerRecord<byte[], byte[]>> records) {
- throw new UnsupportedOperationException("Do not use for TimeOrderedKeyValueStore");
+ final Map<KeyValueSegment, WriteBatch> writeBatchMap = new HashMap<>();
+ for (final ConsumerRecord<byte[], byte[]> record : records) {
+ final long timestamp = WindowKeySchema.extractStoreTimestamp(record.key());
+ observedStreamTime = Math.max(observedStreamTime, timestamp);
+ final long segmentId = segments.segmentId(timestamp);
+ final KeyValueSegment segment = segments.getOrCreateSegmentIfLive(segmentId, context, observedStreamTime);
+ if (segment != null) {
+ //null segment is if it has expired, so we don't want those records
+ ChangelogRecordDeserializationHelper.applyChecksAndUpdatePosition(
+ record,
+ consistencyEnabled,
+ position
+ );
+ try {
+ final WriteBatch batch = writeBatchMap.computeIfAbsent(segment, s -> new WriteBatch());
+
+ final byte[] baseKey = TimeFirstWindowKeySchema.fromNonPrefixWindowKey(record.key());
+ segment.addToBatch(new KeyValue<>(baseKey, record.value()), batch);
+ } catch (final RocksDBException e) {
+ throw new ProcessorStateException("Error restoring batch to store " + name(), e);
+ }
+ }
+ }
+ return writeBatchMap;
}
@Override
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/JoinGracePeriodDurabilityIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/JoinGracePeriodDurabilityIntegrationTest.java
new file mode 100644
index 00000000000..febafff17c5
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/JoinGracePeriodDurabilityIntegrationTest.java
@@ -0,0 +1,256 @@
+/*
+ * 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.streams.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValueTimestamp;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Joined;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.MockValueJoiner;
+import org.apache.kafka.test.TestUtils;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.junit.rules.Timeout;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static java.util.Arrays.asList;
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
+import static org.apache.kafka.common.utils.Utils.mkProperties;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.cleanStateBeforeTest;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.getStartedStreams;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.quietlyCleanStateAfterTest;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+@RunWith(Parameterized.class)
+@Category({IntegrationTest.class})
+public class JoinGracePeriodDurabilityIntegrationTest {
+ @Rule
+ public Timeout globalTimeout = Timeout.seconds(600);
+
+ public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
+ 3,
+ mkProperties(mkMap()),
+ 0L
+ );
+
+ @BeforeClass
+ public static void startCluster() throws IOException {
+ CLUSTER.start();
+ }
+
+ @AfterClass
+ public static void closeCluster() {
+ CLUSTER.stop();
+ }
+
+ @Rule
+ public TestName testName = new TestName();
+
+ private static final StringDeserializer STRING_DESERIALIZER = new StringDeserializer();
+ private static final StringSerializer STRING_SERIALIZER = new StringSerializer();
+ private static final Serde<String> STRING_SERDE = Serdes.String();
+ private static final LongDeserializer LONG_DESERIALIZER = new LongDeserializer();
+ private static final long COMMIT_INTERVAL = 100L;
+
+ @SuppressWarnings("deprecation")
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<String[]> data() {
+ return Arrays.asList(new String[][] {
+ {StreamsConfig.AT_LEAST_ONCE},
+ {StreamsConfig.EXACTLY_ONCE},
+ {StreamsConfig.EXACTLY_ONCE_V2}
+ });
+ }
+
+ @Parameterized.Parameter
+ public String processingGuaranteee;
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void shouldRecoverBufferAfterShutdown() {
+ final String testId = safeUniqueTestName(getClass(), testName);
+ final String appId = "appId_" + testId;
+ final String streamInput = "Streaminput" + testId;
+ final String tableInput = "Tableinput" + testId;
+ final String storeName = "grace";
+ final String output = "output" + testId;
+
+ // change this once task bug fixed.
+ // create multiple partitions as a trap, in case the buffer doesn't properly set the
+ // partition on the records, but instead relies on the default key partitioner
+ cleanStateBeforeTest(CLUSTER, 1, streamInput, tableInput, output);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ final KStream<String, String> stream = builder.stream(streamInput, Consumed.with(STRING_SERDE, STRING_SERDE));
+ final KTable<String, String> table = builder.table(tableInput, Consumed.with(STRING_SERDE, STRING_SERDE), Materialized.as(
+ Stores.persistentVersionedKeyValueStore(storeName, Duration.ofMillis(1000))));
+ final KStream<String, String> joinedStream = stream.join(table,
+ MockValueJoiner.TOSTRING_JOINER,
+ Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "Grace", Duration.ofMillis(5))
+ );
+
+ final AtomicInteger eventCount = new AtomicInteger(0);
+ joinedStream.foreach((key, value) -> eventCount.incrementAndGet());
+
+ joinedStream.to(output);
+
+ final Properties streamsConfig = mkObjectProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
+ mkEntry(StreamsConfig.POLL_MS_CONFIG, Long.toString(COMMIT_INTERVAL)),
+ mkEntry(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuaranteee),
+ mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()),
+ mkEntry(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class),
+ mkEntry(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class)
+ ));
+
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, COMMIT_INTERVAL);
+
+ KafkaStreams driver = getStartedStreams(streamsConfig, builder, true);
+ try {
+ produceSynchronouslyToPartitionZero(
+ tableInput,
+ asList(
+ new KeyValueTimestamp<>("k1", "v1", scaledTime(0L)),
+ new KeyValueTimestamp<>("k2", "v2", scaledTime(0L)),
+ new KeyValueTimestamp<>("k3", "v3", scaledTime(0L)),
+ new KeyValueTimestamp<>("k4", "v4", scaledTime(0L)),
+ new KeyValueTimestamp<>("k5", "v5", scaledTime(0L))
+ )
+ );
+ produceSynchronouslyToPartitionZero(
+ streamInput,
+ asList(
+ new KeyValueTimestamp<>("k1", "v1", scaledTime(1L)),
+ new KeyValueTimestamp<>("k2", "v2", scaledTime(2L)),
+ new KeyValueTimestamp<>("k3", "v3", scaledTime(7L))
+ )
+ );
+ verifyOutput(
+ output,
+ asList(
+ new KeyValueTimestamp<>("k1", "v1+v1", scaledTime(1L)),
+ new KeyValueTimestamp<>("k2", "v2+v2", scaledTime(2L))
+ )
+ );
+ assertThat(eventCount.get(), is(2));
+
+ produceSynchronouslyToPartitionZero(
+ streamInput,
+ asList(
+ new KeyValueTimestamp<>("k4", "v4", scaledTime(4L)),
+ new KeyValueTimestamp<>("k5", "v5", scaledTime(5L))
+ )
+ );
+
+ // bounce to ensure that the history, including retractions,
+ // get restored properly. (i.e., we shouldn't see those first events again)
+
+ // restart the driver
+ driver.close();
+ assertThat(driver.state(), is(KafkaStreams.State.NOT_RUNNING));
+ driver = getStartedStreams(streamsConfig, builder, false);
+
+
+ // flush those recovered buffered events out.
+ produceSynchronouslyToPartitionZero(
+ streamInput,
+ asList(
+ new KeyValueTimestamp<>("k6", "v6", scaledTime(20L))
+ )
+ );
+ verifyOutput(
+ output,
+ asList(
+ new KeyValueTimestamp<>("k4", "v4+v4", scaledTime(4L)),
+ new KeyValueTimestamp<>("k5", "v5+v5", scaledTime(5L))
+ )
+ );
+ assertThat("There should only be 4 output events.", eventCount.get(), is(4));
+
+ } finally {
+ driver.close();
+ quietlyCleanStateAfterTest(CLUSTER, driver);
+ }
+ }
+
+ private void verifyOutput(final String topic, final List<KeyValueTimestamp<String, String>> keyValueTimestamps) {
+ final Properties properties = mkProperties(
+ mkMap(
+ mkEntry(ConsumerConfig.GROUP_ID_CONFIG, "test-group"),
+ mkEntry(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
+ mkEntry(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ((Deserializer<String>) STRING_DESERIALIZER).getClass().getName()),
+ mkEntry(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ((Deserializer<String>) STRING_DESERIALIZER).getClass().getName())
+ )
+ );
+ IntegrationTestUtils.verifyKeyValueTimestamps(properties, topic, keyValueTimestamps);
+ }
+
+ /**
+ * scaling to ensure that there are commits in between the various test events,
+ * just to exercise that everything works properly in the presence of commits.
+ */
+ private long scaledTime(final long unscaledTime) {
+ return COMMIT_INTERVAL * 2 * unscaledTime;
+ }
+
+ private static void produceSynchronouslyToPartitionZero(final String topic, final List<KeyValueTimestamp<String, String>> toProduce) {
+ final Properties producerConfig = mkProperties(mkMap(
+ mkEntry(ProducerConfig.CLIENT_ID_CONFIG, "anything"),
+ mkEntry(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ((Serializer<String>) STRING_SERIALIZER).getClass().getName()),
+ mkEntry(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ((Serializer<String>) STRING_SERIALIZER).getClass().getName()),
+ mkEntry(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers())
+ ));
+ IntegrationTestUtils.produceSynchronously(producerConfig, false, topic, Optional.of(0), toProduce);
+ }
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java
index 3ade7c5e59e..99bef27bf79 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java
@@ -68,7 +68,7 @@ public class RocksDBTimeOrderedKeyValueBufferTest {
private void createBuffer(final Duration grace) {
final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDBTimeOrderedKeyValueBytesStoreSupplier("testing").get();
- buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing");
+ buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing", false);
buffer.setSerdesIfNull(serdeGetter);
buffer.init((StateStoreContext) context, store);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreTest.java
new file mode 100644
index 00000000000..9908dfdb147
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.streams.state.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.MockRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.rocksdb.WriteBatch;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+class RocksDBTimeOrderedKeyValueBytesStoreTest {
+
+ private InternalMockProcessorContext context;
+ private RocksDBTimeOrderedKeyValueBytesStore bytesStore;
+ private File stateDir;
+ final String storeName = "bytes-store";
+ private final static String METRICS_SCOPE = "metrics-scope";
+ private final String topic = "changelog";
+
+
+ @BeforeEach
+ public void before() {
+ bytesStore = new RocksDBTimeOrderedKeyValueBytesStore(storeName, METRICS_SCOPE);
+
+ stateDir = TestUtils.tempDirectory();
+ context = new InternalMockProcessorContext<>(
+ stateDir,
+ Serdes.String(),
+ Serdes.Long(),
+ new MockRecordCollector(),
+ new ThreadCache(new LogContext("testCache "), 0, new MockStreamsMetrics(new Metrics()))
+ );
+ bytesStore.init((StateStoreContext) context, bytesStore);
+ }
+
+ @AfterEach
+ public void close() {
+ bytesStore.close();
+ }
+
+ @Test
+ public void shouldCreateWriteBatches() {
+ final String key = "a";
+ final Collection<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
+ records.add(new ConsumerRecord<>("", 0, 0L, serializeKey(key, 0, 0L).get(), serializeValue(50L)));
+ records.add(new ConsumerRecord<>("", 0, 0L, serializeKey(key, 1, 1L).get(), serializeValue(100L)));
+ final Map<KeyValueSegment, WriteBatch> writeBatchMap = bytesStore.getWriteBatches(records);
+ assertEquals(1, writeBatchMap.size());
+
+ for (final WriteBatch batch : writeBatchMap.values()) {
+ assertEquals(2, batch.count());
+ }
+ }
+
+ @Test
+ public void shouldCreateEmptyWriteBatches() {
+ final Collection<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
+ final Map<KeyValueSegment, WriteBatch> writeBatchMap = bytesStore.getWriteBatches(records);
+ assertEquals(0, writeBatchMap.size());
+ }
+
+ private byte[] serializeValue(final Long value) {
+ final Serde<Long> valueSerde = new Serdes.LongSerde();
+ final byte[] valueBytes = valueSerde.serializer().serialize(topic, value);
+ final BufferValue buffered = new BufferValue(null, null, valueBytes, new ProcessorRecordContext(0, 0, 0, topic, new RecordHeaders()));
+ return buffered.serialize(0).array();
+ }
+
+ private Bytes serializeKey(final String key, final int seqnum, final long timestamp) {
+ final Serde<String> keySerde = new Serdes.StringSerde();
+ return Bytes.wrap(
+ PrefixedWindowKeySchemas.TimeFirstWindowKeySchema.toStoreKeyBinary(keySerde.serializer().serialize(topic, key),
+ timestamp,
+ seqnum).get());
+ }
+}
\ No newline at end of file