From b652b087e769fd9cfc019f5ae7f5874aed97b073 Mon Sep 17 00:00:00 2001 From: Josep Prat Date: Wed, 19 May 2021 14:06:58 +0200 Subject: [PATCH 1/3] KAFKA-12809: Remove Deprecated methods under Stores Removes deprecated methods since 2.1 Moves needed implementation from deprecated method to the right new one. --- .../apache/kafka/streams/state/Stores.java | 74 +++---------------- .../kafka/streams/state/StoresTest.java | 7 -- 2 files changed, 9 insertions(+), 72 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java index 01016b729b9b6..7b088270082fa 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java @@ -175,46 +175,6 @@ public String metricsScope() { }; } - /** - * Create a persistent {@link WindowBytesStoreSupplier}. - * - * @param name name of the store (cannot be {@code null}) - * @param retentionPeriod length of time to retain data in the store (cannot be negative) - * (note that the retention period must be at least long enough to contain the - * windowed data's entire life cycle, from window-start through window-end, - * and for the entire grace period) - * @param numSegments number of db segments (cannot be zero or negative) - * @param windowSize size of the windows that are stored (cannot be negative). Note: the window size - * is not stored with the records, so this value is used to compute the keys that - * the store returns. No effort is made to validate this parameter, so you must be - * careful to set it the same as the windowed keys you're actually storing. - * @param retainDuplicates whether or not to retain duplicates. Turning this on will automatically disable - * caching and means that null values will be ignored. - * @return an instance of {@link WindowBytesStoreSupplier} - * @deprecated since 2.1 Use {@link Stores#persistentWindowStore(String, Duration, Duration, boolean)} instead - */ - @Deprecated // continuing to support Windows#maintainMs/segmentInterval in fallback mode - public static WindowBytesStoreSupplier persistentWindowStore(final String name, - final long retentionPeriod, - final int numSegments, - final long windowSize, - final boolean retainDuplicates) { - if (numSegments < 2) { - throw new IllegalArgumentException("numSegments cannot be smaller than 2"); - } - - final long legacySegmentInterval = Math.max(retentionPeriod / (numSegments - 1), 60_000L); - - return persistentWindowStore( - name, - retentionPeriod, - windowSize, - retainDuplicates, - legacySegmentInterval, - false - ); - } - /** * Create a persistent {@link WindowBytesStoreSupplier}. *

@@ -360,26 +320,6 @@ public static WindowBytesStoreSupplier inMemoryWindowStore(final String name, return new InMemoryWindowBytesStoreSupplier(name, retentionMs, windowSizeMs, retainDuplicates); } - /** - * Create a persistent {@link SessionBytesStoreSupplier}. - * - * @param name name of the store (cannot be {@code null}) - * @param retentionPeriodMs length of time to retain data in the store (cannot be negative) - * (note that the retention period must be at least as long enough to - * contain the inactivity gap of the session and the entire grace period.) - * @return an instance of a {@link SessionBytesStoreSupplier} - * @deprecated since 2.1 Use {@link Stores#persistentSessionStore(String, Duration)} instead - */ - @Deprecated // continuing to support Windows#maintainMs/segmentInterval in fallback mode - public static SessionBytesStoreSupplier persistentSessionStore(final String name, - final long retentionPeriodMs) { - Objects.requireNonNull(name, "name cannot be null"); - if (retentionPeriodMs < 0) { - throw new IllegalArgumentException("retentionPeriod cannot be negative"); - } - return new RocksDbSessionBytesStoreSupplier(name, retentionPeriodMs); - } - /** * Create a persistent {@link SessionBytesStoreSupplier}. * @@ -389,11 +329,15 @@ public static SessionBytesStoreSupplier persistentSessionStore(final String name * contain the inactivity gap of the session and the entire grace period.) * @return an instance of a {@link SessionBytesStoreSupplier} */ - @SuppressWarnings("deprecation") // removing #persistentSessionStore(String name, long retentionPeriodMs) will fix this - public static SessionBytesStoreSupplier persistentSessionStore(final String name, - final Duration retentionPeriod) { - final String msgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod"); - return persistentSessionStore(name, validateMillisecondDuration(retentionPeriod, msgPrefix)); + public static SessionBytesStoreSupplier persistentSessionStore(final String name, + final Duration retentionPeriod) { + final String msgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod"); + long retentionPeriodMs = validateMillisecondDuration(retentionPeriod, msgPrefix); + Objects.requireNonNull(name, "name cannot be null"); + if (retentionPeriodMs < 0) { + throw new IllegalArgumentException("retentionPeriod cannot be negative"); + } + return new RocksDbSessionBytesStoreSupplier(name, retentionPeriodMs); } /** diff --git a/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java b/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java index c4adccdc3f175..90f019aa7ccbe 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java @@ -95,13 +95,6 @@ public void shouldThrowIfIPersistentTimestampedWindowStoreRetentionPeriodIsNegat assertEquals("retentionPeriod cannot be negative", e.getMessage()); } - @Deprecated - @Test - public void shouldThrowIfIPersistentWindowStoreIfNumberOfSegmentsSmallerThanOne() { - final Exception e = assertThrows(IllegalArgumentException.class, () -> Stores.persistentWindowStore("anyName", 0L, 1, 0L, false)); - assertEquals("numSegments cannot be smaller than 2", e.getMessage()); - } - @Test public void shouldThrowIfIPersistentWindowStoreIfWindowSizeIsNegative() { final Exception e = assertThrows(IllegalArgumentException.class, () -> Stores.persistentWindowStore("anyName", ofMillis(0L), ofMillis(-1L), false)); From 66ef60c910639271728e7a4f670c696d3f6facc0 Mon Sep 17 00:00:00 2001 From: Josep Prat Date: Wed, 19 May 2021 14:31:49 +0200 Subject: [PATCH 2/3] Fix indentation --- .../org/apache/kafka/streams/state/Stores.java | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java index 7b088270082fa..f623eaa72d855 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java @@ -329,15 +329,15 @@ public static WindowBytesStoreSupplier inMemoryWindowStore(final String name, * contain the inactivity gap of the session and the entire grace period.) * @return an instance of a {@link SessionBytesStoreSupplier} */ - public static SessionBytesStoreSupplier persistentSessionStore(final String name, - final Duration retentionPeriod) { - final String msgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod"); - long retentionPeriodMs = validateMillisecondDuration(retentionPeriod, msgPrefix); - Objects.requireNonNull(name, "name cannot be null"); - if (retentionPeriodMs < 0) { - throw new IllegalArgumentException("retentionPeriod cannot be negative"); - } - return new RocksDbSessionBytesStoreSupplier(name, retentionPeriodMs); + public static SessionBytesStoreSupplier persistentSessionStore(final String name, + final Duration retentionPeriod) { + final String msgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod"); + final long retentionPeriodMs = validateMillisecondDuration(retentionPeriod, msgPrefix); + Objects.requireNonNull(name, "name cannot be null"); + if (retentionPeriodMs < 0) { + throw new IllegalArgumentException("retentionPeriod cannot be negative"); + } + return new RocksDbSessionBytesStoreSupplier(name, retentionPeriodMs); } /** From da9c2445d25e4d90cae1a00065075c82b613e382 Mon Sep 17 00:00:00 2001 From: Josep Prat Date: Wed, 19 May 2021 20:56:20 +0200 Subject: [PATCH 3/3] Do null check as first operation --- .../src/main/java/org/apache/kafka/streams/state/Stores.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java index f623eaa72d855..12df0e01dffed 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java @@ -331,9 +331,9 @@ public static WindowBytesStoreSupplier inMemoryWindowStore(final String name, */ public static SessionBytesStoreSupplier persistentSessionStore(final String name, final Duration retentionPeriod) { + Objects.requireNonNull(name, "name cannot be null"); final String msgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod"); final long retentionPeriodMs = validateMillisecondDuration(retentionPeriod, msgPrefix); - Objects.requireNonNull(name, "name cannot be null"); if (retentionPeriodMs < 0) { throw new IllegalArgumentException("retentionPeriod cannot be negative"); }