MINOR: buffer should ignore caching

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,14 +69,28 @@ 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.
+         *
+         * It's currently a no-op.
+         */
         @Override
         public StoreBuilder<StateStore> withCachingEnabled() {
-            throw new UnsupportedOperationException();
+            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.
+         *
+         * It's currently a no-op.
+         */
         @Override
         public StoreBuilder<StateStore> withCachingDisabled() {
-            throw new UnsupportedOperationException();
+            return this;
         }
 
         @Override

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();
+    }
+}