You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by bb...@apache.org on 2026/08/11 17:06:27 UTC
(kafka) branch trunk updated: KAFKA-20875: Headers window store adapters must expose the retention they wrap (#23037)
This is an automated email from the ASF dual-hosted git repository.
bbejeck 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 e69bbcdeaa9 KAFKA-20875: Headers window store adapters must expose the retention they wrap (#23037)
e69bbcdeaa9 is described below
commit e69bbcdeaa9ff83b9b5e768c318109a9ee497976
Author: Alan Lau <72...@users.noreply.github.com>
AuthorDate: Tue Aug 11 13:06:18 2026 -0400
KAFKA-20875: Headers window store adapters must expose the retention they wrap (#23037)
issue: https://issues.apache.org/jira/browse/KAFKA-20875
`PlainToHeadersWindowStoreAdapter` and
`TimestampedToHeadersWindowStoreAdapter` hold the store they wrap in a
field rather than participating in the `WrappedStateStore` chain, so
`ProcessorStateManager.StateStoreMetadata.extractRetentionPeriod` —
which finds a store's retention by unwrapping to the innermost layer —
terminates on the adapter and resolves -1.
A -1 fails the windowed restore gate `retentionPeriod > 0 &&
retentionPeriod != Long.MAX_VALUE`, so the restore seeks the beginning
of the changelog rather than skipping data the store would discard
anyway. On a changelog with delete retention that means starting at the
offset retention is deleting from, with no margin against being lapped.
Each adapter now implements `WithRetentionPeriod` and unwraps its own
delegate.
Reviewers: Bill Bejeck <bb...@apache.org>
---
.../processor/internals/ProcessorStateManager.java | 14 +---
.../processor/internals/StoreChangelogReader.java | 9 +++
.../PlainToHeadersWindowStoreAdapter.java | 13 +++-
.../internals/SessionToHeadersStoreAdapter.java | 9 ++-
.../TimestampedToHeadersWindowStoreAdapter.java | 13 +++-
.../WindowToTimestampedWindowByteStoreAdapter.java | 9 ++-
.../state/internals/WithRetentionPeriod.java | 17 +++++
.../PlainToHeadersWindowStoreAdapterTest.java | 8 ++
.../SessionToHeadersStoreAdapterTest.java | 17 +++++
...TimestampedToHeadersWindowStoreAdapterTest.java | 8 ++
.../TimestampedWindowStoreBuilderTest.java | 27 +++++++
.../state/internals/WithRetentionPeriodTest.java | 88 ++++++++++++++++++++++
12 files changed, 215 insertions(+), 17 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
index f6ec2ee4556..21a52e64448 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
@@ -38,7 +38,6 @@ import org.apache.kafka.streams.state.internals.LegacyCheckpointingStateStore;
import org.apache.kafka.streams.state.internals.RecordConverter;
import org.apache.kafka.streams.state.internals.TimeOrderedKeyValueBuffer;
import org.apache.kafka.streams.state.internals.WithRetentionPeriod;
-import org.apache.kafka.streams.state.internals.WrappedStateStore;
import org.slf4j.Logger;
@@ -137,24 +136,13 @@ public class ProcessorStateManager implements StateManager {
this.commitCallback = commitCallback;
this.recordConverter = recordConverter;
this.offset = null;
- this.retentionPeriod = extractRetentionPeriod(stateStore);
+ this.retentionPeriod = WithRetentionPeriod.resolveRetentionPeriod(stateStore);
}
private void setOffset(final Long offset) {
this.offset = offset;
}
- private static long extractRetentionPeriod(final StateStore stateStore) {
- StateStore current = stateStore;
- while (current instanceof WrappedStateStore) {
- current = ((WrappedStateStore<?, ?, ?>) current).wrapped();
- }
- if (current instanceof WithRetentionPeriod) {
- return ((WithRetentionPeriod) current).retentionPeriod();
- }
- return -1L;
- }
-
// the offset is exposed to the changelog reader to determine if restoration is completed
Long offset() {
return this.offset;
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
index d45d6b50a49..9c5093081b1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
@@ -38,9 +38,12 @@ import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskCorruptedException;
import org.apache.kafka.streams.processor.StandbyUpdateListener;
import org.apache.kafka.streams.processor.StateRestoreListener;
+import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.ProcessorStateManager.StateStoreMetadata;
import org.apache.kafka.streams.processor.internals.Task.TaskType;
+import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.internals.MeteredStateStore;
import org.slf4j.Logger;
@@ -1035,6 +1038,12 @@ public class StoreChangelogReader implements ChangelogReader {
if (retentionPeriod > 0 && retentionPeriod != Long.MAX_VALUE) {
newWindowedPartitionsRetention.put(partition, retentionPeriod);
} else {
+ final StateStore store = storeMetadata.store();
+ if (store instanceof WindowStore || store instanceof SessionStore) {
+ log.warn("Windowed store {} reported no usable retention period ({}), so changelog " +
+ "partition {} is restored in full rather than skipping expired data.",
+ store.name(), retentionPeriod, partition);
+ }
log.debug("Start restoring changelog partition {} from the beginning offset to end offset {} " +
"since we cannot find current offset.", partition, recordEndOffset(endOffset));
newSeekToBeginningPartitions.add(partition);
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java
index 218182878e8..72edeb866c8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java
@@ -55,7 +55,7 @@ import static org.apache.kafka.streams.state.internals.Utils.rawPlainValue;
* <li>Read: {@code [value]} → {@code [headers][timestamp][value]} (add empty headers and timestamp=-1)</li>
* </ul>
*/
-public class PlainToHeadersWindowStoreAdapter implements WindowStore<Bytes, byte[]> {
+public class PlainToHeadersWindowStoreAdapter implements WindowStore<Bytes, byte[]>, WithRetentionPeriod {
private final WindowStore<Bytes, byte[]> store;
public PlainToHeadersWindowStoreAdapter(final WindowStore<Bytes, byte[]> store) {
@@ -68,6 +68,17 @@ public class PlainToHeadersWindowStoreAdapter implements WindowStore<Bytes, byte
this.store = store;
}
+ /**
+ * Report the retention of the store being adapted. This adapter holds its delegate in a private
+ * field rather than as a {@link WrappedStateStore}, so {@code extractRetentionPeriod}'s unwrap
+ * walk terminates here; without this it resolves -1 and the KAFKA-13499 windowed-restore
+ * optimisation is silently skipped for every store behind the adapter.
+ */
+ @Override
+ public long retentionPeriod() {
+ return WithRetentionPeriod.resolveRetentionPeriod(store);
+ }
+
@Override
public void put(final Bytes key, final byte[] valueWithTimestampAndHeaders, final long windowStartTimestamp) {
store.put(key, rawPlainValue(valueWithTimestampAndHeaders), windowStartTimestamp);
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java
index 91c79234325..6eb8e858294 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java
@@ -53,9 +53,16 @@ import static org.apache.kafka.streams.state.HeadersBytesStore.convertToHeaderFo
* @see SessionToHeadersIteratorAdapter
*/
@SuppressWarnings("unchecked")
-public class SessionToHeadersStoreAdapter implements SessionStore<Bytes, byte[]> {
+public class SessionToHeadersStoreAdapter implements SessionStore<Bytes, byte[]>, WithRetentionPeriod {
final SessionStore<Bytes, byte[]> store;
+ // the delegate is held in a field rather than as a WrappedStateStore, so an unwrap walk
+ // terminates here; without this the windowed restore optimisation resolves -1 and is skipped
+ @Override
+ public long retentionPeriod() {
+ return WithRetentionPeriod.resolveRetentionPeriod(store);
+ }
+
SessionToHeadersStoreAdapter(final SessionStore<Bytes, byte[]> store) {
if (!store.persistent()) {
throw new IllegalArgumentException("Provided store must be a persistent store, but it is not.");
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
index 889533f5534..880da9b3b11 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
@@ -56,9 +56,20 @@ import static org.apache.kafka.streams.state.internals.Utils.rawTimestampedValue
* <li>Read: {@code [timestamp][value]} → {@code [headers][timestamp][value]} (add empty headers)</li>
* </ul>
*/
-public class TimestampedToHeadersWindowStoreAdapter implements WindowStore<Bytes, byte[]> {
+public class TimestampedToHeadersWindowStoreAdapter implements WindowStore<Bytes, byte[]>, WithRetentionPeriod {
final WindowStore<Bytes, byte[]> store;
+ /**
+ * Report the retention of the store being adapted. This adapter holds its delegate in a field
+ * rather than as a {@link WrappedStateStore}, so {@code extractRetentionPeriod}'s unwrap walk
+ * terminates here; without this it resolves -1 and the windowed restore optimisation is
+ * silently skipped for every store behind the adapter.
+ */
+ @Override
+ public long retentionPeriod() {
+ return WithRetentionPeriod.resolveRetentionPeriod(store);
+ }
+
public TimestampedToHeadersWindowStoreAdapter(final WindowStore<Bytes, byte[]> store) {
if (!store.persistent()) {
throw new IllegalArgumentException("Provided store must be a persistent store, but it is not.");
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
index eb037c77534..38de76a0b3a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
@@ -36,9 +36,16 @@ import java.util.Map;
import static org.apache.kafka.streams.state.TimestampedBytesStore.convertToTimestampedFormat;
import static org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.rawValue;
-public class WindowToTimestampedWindowByteStoreAdapter implements WindowStore<Bytes, byte[]> {
+public class WindowToTimestampedWindowByteStoreAdapter implements WindowStore<Bytes, byte[]>, WithRetentionPeriod {
final WindowStore<Bytes, byte[]> store;
+ // the delegate is held in a field rather than as a WrappedStateStore, so an unwrap walk
+ // terminates here; without this the windowed restore optimisation resolves -1 and is skipped
+ @Override
+ public long retentionPeriod() {
+ return WithRetentionPeriod.resolveRetentionPeriod(store);
+ }
+
WindowToTimestampedWindowByteStoreAdapter(final WindowStore<Bytes, byte[]> store) {
if (!store.persistent()) {
throw new IllegalArgumentException("Provided store must be a persistent store, but it is not.");
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WithRetentionPeriod.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WithRetentionPeriod.java
index 44dee28eafa..ba3ef5237fd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/WithRetentionPeriod.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WithRetentionPeriod.java
@@ -16,6 +16,23 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.streams.processor.StateStore;
+
public interface WithRetentionPeriod {
long retentionPeriod();
+
+ /**
+ * Unwrap to the innermost store and return its retention, or -1 if it does not report one.
+ * Stores that hold their delegate in a field rather than as a {@link WrappedStateStore} end the
+ * walk, so they resolve their own delegate through here.
+ */
+ static long resolveRetentionPeriod(final StateStore store) {
+ StateStore current = store;
+ while (current instanceof WrappedStateStore) {
+ current = ((WrappedStateStore<?, ?, ?>) current).wrapped();
+ }
+ return current instanceof WithRetentionPeriod
+ ? ((WithRetentionPeriod) current).retentionPeriod()
+ : -1L;
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java
index 24bc67dc00e..babf8a11330 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java
@@ -300,4 +300,12 @@ public class PlainToHeadersWindowStoreAdapterTest {
}
assertFalse(foundAdapterInfo, "Expected no execution info from adapter when not requested");
}
+
+ @Test
+ public void shouldReportRetentionPeriodOfUnderlyingStore() {
+ // typed as StateStore so this compiles, and fails, when the adapter does not expose it
+ final StateStore store = adapter;
+ assertInstanceOf(WithRetentionPeriod.class, store);
+ assertEquals(RETENTION_PERIOD, ((WithRetentionPeriod) store).retentionPeriod());
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapterTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapterTest.java
index 2f3b3ebe65f..6330ff38be5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapterTest.java
@@ -48,6 +48,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.STRICT_STUBS)
@@ -253,4 +254,20 @@ public class SessionToHeadersStoreAdapterTest {
public void shouldReturnNullFromRawAggregationValueForNull() {
assertNull(Utils.rawAggregation(null));
}
+
+ @Test
+ public void shouldReportRetentionPeriodOfUnderlyingStore() {
+ final long retentionMs = 60_000L;
+ @SuppressWarnings("unchecked")
+ final SessionStore<Bytes, byte[]> retentionAwareStore =
+ mock(SessionStore.class, withSettings().extraInterfaces(WithRetentionPeriod.class));
+ when(retentionAwareStore.persistent()).thenReturn(true);
+ when(((WithRetentionPeriod) retentionAwareStore).retentionPeriod()).thenReturn(retentionMs);
+
+ // typed as StateStore so this compiles, and fails, when the adapter does not expose it
+ final StateStore store = new SessionToHeadersStoreAdapter(retentionAwareStore);
+
+ assertInstanceOf(WithRetentionPeriod.class, store);
+ assertEquals(retentionMs, ((WithRetentionPeriod) store).retentionPeriod());
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java
index 4b757785c42..e03315b64bc 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java
@@ -319,4 +319,12 @@ public class TimestampedToHeadersWindowStoreAdapterTest {
}
assertFalse(foundAdapterInfo, "Expected no execution info from adapter when not requested");
}
+
+ @Test
+ public void shouldReportRetentionPeriodOfUnderlyingStore() {
+ // typed as StateStore so this compiles, and fails, when the adapter does not expose it
+ final StateStore store = adapter;
+ assertInstanceOf(WithRetentionPeriod.class, store);
+ assertEquals(RETENTION_PERIOD, ((WithRetentionPeriod) store).retentionPeriod());
+ }
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedWindowStoreBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedWindowStoreBuilderTest.java
index c403614e2b7..ddaedbe0a50 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedWindowStoreBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedWindowStoreBuilderTest.java
@@ -39,6 +39,7 @@ import java.time.Duration;
import java.util.Collections;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -198,6 +199,32 @@ public class TimestampedWindowStoreBuilderTest {
assertThat(((WrappedStateStore) store).wrapped(), instanceOf(WindowToTimestampedWindowByteStoreAdapter.class));
}
+ @ValueSource(strings = {TIMESTAMP_STORE_NAME, TIMEORDERED_STORE_NAME})
+ @ParameterizedTest
+ public void shouldExposeRetentionPeriodThroughTimestampedAdapter(final String storeName) {
+ final long retentionMs = 10L;
+ setUp(storeName);
+ when(supplier.get()).thenReturn(new RocksDBWindowStore(
+ new RocksDBSegmentedBytesStore(
+ "name",
+ "metric-scope",
+ retentionMs,
+ 5L,
+ new WindowKeySchema()),
+ false,
+ 1L));
+
+ final TimestampedWindowStore<String, String> store = builder
+ .withLoggingDisabled()
+ .withCachingDisabled()
+ .build();
+
+ // typed as StateStore so this compiles, and fails, when the adapter does not expose it
+ final StateStore adapter = ((WrappedStateStore) store).wrapped();
+ assertThat(adapter, instanceOf(WithRetentionPeriod.class));
+ assertThat(((WithRetentionPeriod) adapter).retentionPeriod(), is(retentionMs));
+ }
+
@ValueSource(strings = {TIMESTAMP_STORE_NAME, TIMEORDERED_STORE_NAME})
@ParameterizedTest
public void shouldDisableCachingWithRetainDuplicates(final String storeName) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/WithRetentionPeriodTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/WithRetentionPeriodTest.java
new file mode 100644
index 00000000000..e6983fad8db
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/WithRetentionPeriodTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.streams.processor.StateStore;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+public class WithRetentionPeriodTest {
+
+ private static final long RETENTION_MS = 60_000L;
+
+ private static class TestWrapper extends WrappedStateStore<StateStore, Object, Object> {
+ TestWrapper(final StateStore wrapped) {
+ super(wrapped);
+ }
+ }
+
+ private static StateStore storeReportingRetention() {
+ final StateStore store = mock(StateStore.class, withSettings().extraInterfaces(WithRetentionPeriod.class));
+ when(((WithRetentionPeriod) store).retentionPeriod()).thenReturn(RETENTION_MS);
+ return store;
+ }
+
+ @Test
+ public void shouldReturnRetentionOfStoreThatReportsItDirectly() {
+ assertEquals(RETENTION_MS, WithRetentionPeriod.resolveRetentionPeriod(storeReportingRetention()));
+ }
+
+ @Test
+ public void shouldResolveRetentionThroughWrappedStores() {
+ final StateStore wrapped = new TestWrapper(new TestWrapper(storeReportingRetention()));
+
+ assertEquals(RETENTION_MS, WithRetentionPeriod.resolveRetentionPeriod(wrapped));
+ }
+
+ @Test
+ public void shouldReturnMinusOneWhenStoreDoesNotReportRetention() {
+ assertEquals(-1L, WithRetentionPeriod.resolveRetentionPeriod(mock(StateStore.class)));
+ }
+
+ @Test
+ public void shouldReturnMinusOneWhenNoWrappedStoreReportsRetention() {
+ final StateStore wrapped = new TestWrapper(new TestWrapper(mock(StateStore.class)));
+
+ assertEquals(-1L, WithRetentionPeriod.resolveRetentionPeriod(wrapped));
+ }
+
+ @Test
+ public void shouldNotConsultOuterLayersThatReportRetention() {
+ // retention belongs to the innermost store, so an outer layer reporting one is ignored
+ final StateStore outer = new RetentionReportingWrapper(mock(StateStore.class));
+
+ assertEquals(-1L, WithRetentionPeriod.resolveRetentionPeriod(outer));
+ }
+
+ private static class RetentionReportingWrapper extends WrappedStateStore<StateStore, Object, Object>
+ implements WithRetentionPeriod {
+
+ RetentionReportingWrapper(final StateStore wrapped) {
+ super(wrapped);
+ }
+
+ @Override
+ public long retentionPeriod() {
+ return RETENTION_MS;
+ }
+ }
+}