From 1dad8093d0b71d34c046a786c53860daeee49e03 Mon Sep 17 00:00:00 2001 From: John Roesler Date: Fri, 19 Oct 2018 15:26:07 -0500 Subject: [PATCH 1/3] MINOR: buffer should ignore caching --- .../state/internals/InMemoryTimeOrderedKeyValueBuffer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java index 3ac6fc87073a2..905a38bb45a4d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java @@ -74,12 +74,12 @@ public Builder(final String storeName) { @Override public StoreBuilder withCachingEnabled() { - throw new UnsupportedOperationException(); + return this; } @Override public StoreBuilder withCachingDisabled() { - throw new UnsupportedOperationException(); + return this; } @Override From 9bdfce8cd9ebc7219e034a352701ec32089d46ea Mon Sep 17 00:00:00 2001 From: John Roesler Date: Sat, 20 Oct 2018 15:08:05 -0500 Subject: [PATCH 2/3] CR: add doc and test --- .../InMemoryTimeOrderedKeyValueBuffer.java | 13 ++++++-- ...InMemoryTimeOrderedKeyValueBufferTest.java | 32 +++++++++++++++++++ 2 files changed, 42 insertions(+), 3 deletions(-) create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBufferTest.java diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java index 905a38bb45a4d..57a9f88d73e9a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java @@ -28,8 +28,6 @@ import org.apache.kafka.streams.processor.internals.RecordBatchingStateRestoreCallback; import org.apache.kafka.streams.processor.internals.RecordCollector; import org.apache.kafka.streams.state.StoreBuilder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; import java.util.Collection; @@ -45,7 +43,6 @@ import java.util.function.Supplier; public class InMemoryTimeOrderedKeyValueBuffer implements TimeOrderedKeyValueBuffer { - private static final Logger LOG = LoggerFactory.getLogger(InMemoryTimeOrderedKeyValueBuffer.class); private static final BytesSerializer KEY_SERIALIZER = new BytesSerializer(); private static final ByteArraySerializer VALUE_SERIALIZER = new ByteArraySerializer(); @@ -72,11 +69,21 @@ public Builder(final String storeName) { this.storeName = storeName; } + /** + * As of 2.1, there's no way for users to directly interact with the buffer, + * so this method is implemented solely to be called by Streams (which + * it will do based on the {@code cache.max.bytes.buffering} config. + */ @Override public StoreBuilder withCachingEnabled() { return this; } + /** + * As of 2.1, there's no way for users to directly interact with the buffer, + * so this method is implemented solely to be called by Streams (which + * it will do based on the {@code cache.max.bytes.buffering} config. + */ @Override public StoreBuilder withCachingDisabled() { return this; diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBufferTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBufferTest.java new file mode 100644 index 0000000000000..ddc40462d4d7b --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBufferTest.java @@ -0,0 +1,32 @@ +/* + * 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.junit.Test; + +public class InMemoryTimeOrderedKeyValueBufferTest { + + @Test + public void bufferShouldAllowCacheEnablement() { + new InMemoryTimeOrderedKeyValueBuffer.Builder(null).withCachingEnabled(); + } + + @Test + public void bufferShouldAllowCacheDisablement() { + new InMemoryTimeOrderedKeyValueBuffer.Builder(null).withCachingDisabled(); + } +} From e6c42b021c4e4181b8cca7c9296f92e0bbdd2107 Mon Sep 17 00:00:00 2001 From: John Roesler Date: Sat, 20 Oct 2018 16:25:02 -0500 Subject: [PATCH 3/3] wording --- .../state/internals/InMemoryTimeOrderedKeyValueBuffer.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java index 57a9f88d73e9a..d94f671b712e4 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java @@ -73,6 +73,8 @@ public Builder(final String storeName) { * As of 2.1, there's no way for users to directly interact with the buffer, * so this method is implemented solely to be called by Streams (which * it will do based on the {@code cache.max.bytes.buffering} config. + * + * It's currently a no-op. */ @Override public StoreBuilder withCachingEnabled() { @@ -83,6 +85,8 @@ public StoreBuilder withCachingEnabled() { * As of 2.1, there's no way for users to directly interact with the buffer, * so this method is implemented solely to be called by Streams (which * it will do based on the {@code cache.max.bytes.buffering} config. + * + * It's currently a no-op. */ @Override public StoreBuilder withCachingDisabled() {