KAFKA-12749: Changelog topic config on suppressed KTable lost

streams/src/main/java/org/apache/kafka/streams/kstream/Suppressed.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.streams.kstream.internals.suppress.SuppressedInternal;
 
 import java.time.Duration;
+import java.util.Collections;
 import java.util.Map;
 
 public interface Suppressed<K> extends NamedOperation<Suppressed<K>> {
@@ -48,7 +49,7 @@ interface BufferConfig<BC extends BufferConfig<BC>> {
          * Create a size-constrained buffer in terms of the maximum number of keys it will store.
          */
         static EagerBufferConfig maxRecords(final long recordLimit) {
-            return new EagerBufferConfigImpl(recordLimit, Long.MAX_VALUE);
+            return new EagerBufferConfigImpl(recordLimit, Long.MAX_VALUE, Collections.emptyMap());
         }
 
         /**
@@ -60,7 +61,7 @@ static EagerBufferConfig maxRecords(final long recordLimit) {
          * Create a size-constrained buffer in terms of the maximum number of bytes it will use.
          */
         static EagerBufferConfig maxBytes(final long byteLimit) {
-            return new EagerBufferConfigImpl(Long.MAX_VALUE, byteLimit);
+            return new EagerBufferConfigImpl(Long.MAX_VALUE, byteLimit, Collections.emptyMap());
         }
 
         /**

streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/BufferConfigInternal.java

@@ -35,18 +35,19 @@ public Suppressed.StrictBufferConfig withNoBound() {
         return new StrictBufferConfigImpl(
             Long.MAX_VALUE,
             Long.MAX_VALUE,
-            SHUT_DOWN // doesn't matter, given the bounds
+            SHUT_DOWN, // doesn't matter, given the bounds
+            getLogConfig()
         );
     }
 
     @Override
     public Suppressed.StrictBufferConfig shutDownWhenFull() {
-        return new StrictBufferConfigImpl(maxRecords(), maxBytes(), SHUT_DOWN);
+        return new StrictBufferConfigImpl(maxRecords(), maxBytes(), SHUT_DOWN, getLogConfig());
     }
 
     @Override
     public Suppressed.EagerBufferConfig emitEarlyWhenFull() {
-        return new EagerBufferConfigImpl(maxRecords(), maxBytes());
+        return new EagerBufferConfigImpl(maxRecords(), maxBytes(), getLogConfig());
     }
 
     public abstract boolean isLoggingEnabled();

streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java

@@ -28,13 +28,7 @@ public class EagerBufferConfigImpl extends BufferConfigInternal<Suppressed.Eager
     private final long maxBytes;
     private final Map<String, String> logConfig;
 
-    public EagerBufferConfigImpl(final long maxRecords, final long maxBytes) {
-        this.maxRecords = maxRecords;
-        this.maxBytes = maxBytes;
-        this.logConfig = Collections.emptyMap();
-    }
-
-    private EagerBufferConfigImpl(final long maxRecords,
+    public EagerBufferConfigImpl(final long maxRecords,
                                   final long maxBytes,
                                   final Map<String, String> logConfig) {
         this.maxRecords = maxRecords;

streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/StrictBufferConfigImpl.java

@@ -41,14 +41,6 @@ public StrictBufferConfigImpl(final long maxRecords,
         this.logConfig = logConfig;
     }
 
-    public StrictBufferConfigImpl(final long maxRecords,
-                                  final long maxBytes,
-                                  final BufferFullStrategy bufferFullStrategy) {
-        this.maxRecords = maxRecords;
-        this.maxBytes = maxBytes;
-        this.bufferFullStrategy = bufferFullStrategy;
-        this.logConfig = Collections.emptyMap();
-    }
 
     public StrictBufferConfigImpl() {
         this.maxRecords = Long.MAX_VALUE;
@@ -59,12 +51,12 @@ public StrictBufferConfigImpl() {
 
     @Override
     public Suppressed.StrictBufferConfig withMaxRecords(final long recordLimit) {
-        return new StrictBufferConfigImpl(recordLimit, maxBytes, bufferFullStrategy);
+        return new StrictBufferConfigImpl(recordLimit, maxBytes, bufferFullStrategy, getLogConfig());
     }
 
     @Override
     public Suppressed.StrictBufferConfig withMaxBytes(final long byteLimit) {
-        return new StrictBufferConfigImpl(maxRecords, byteLimit, bufferFullStrategy);
+        return new StrictBufferConfigImpl(maxRecords, byteLimit, bufferFullStrategy, getLogConfig());
     }
 
     @Override

streams/src/test/java/org/apache/kafka/streams/kstream/SuppressedTest.java

@@ -22,6 +22,8 @@
 import org.apache.kafka.streams.kstream.internals.suppress.SuppressedInternal;
 import org.junit.Test;
 
+import java.util.Collections;
+
 import static java.lang.Long.MAX_VALUE;
 import static java.time.Duration.ofMillis;
 import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxBytes;
@@ -46,13 +48,13 @@ public void bufferBuilderShouldBeConsistent() {
         assertThat(
             "keys alone should be set",
             maxRecords(2L),
-            is(new EagerBufferConfigImpl(2L, MAX_VALUE))
+            is(new EagerBufferConfigImpl(2L, MAX_VALUE, Collections.emptyMap()))
         );
 
         assertThat(
             "size alone should be set",
             maxBytes(2L),
-            is(new EagerBufferConfigImpl(MAX_VALUE, 2L))
+            is(new EagerBufferConfigImpl(MAX_VALUE, 2L, Collections.emptyMap()))
         );
     }
 
@@ -91,7 +93,7 @@ public void intermediateEventsShouldAcceptAnyBufferAndSetBounds() {
         assertThat(
             "all constraints should be set",
             untilTimeLimit(ofMillis(2L), maxRecords(3L).withMaxBytes(2L)),
-            is(new SuppressedInternal<>(null, ofMillis(2), new EagerBufferConfigImpl(3L, 2L), null, false))
+            is(new SuppressedInternal<>(null, ofMillis(2), new EagerBufferConfigImpl(3L, 2L, Collections.emptyMap()), null, false))
         );
     }
 
@@ -105,13 +107,13 @@ public void finalEventsShouldAcceptStrictBuffersAndSetBounds() {
 
         assertThat(
             untilWindowCloses(maxRecords(2L).shutDownWhenFull()),
-            is(new FinalResultsSuppressionBuilder<>(null, new StrictBufferConfigImpl(2L, MAX_VALUE, SHUT_DOWN))
+            is(new FinalResultsSuppressionBuilder<>(null, new StrictBufferConfigImpl(2L, MAX_VALUE, SHUT_DOWN, Collections.emptyMap()))
             )
         );
 
         assertThat(
             untilWindowCloses(maxBytes(2L).shutDownWhenFull()),
-            is(new FinalResultsSuppressionBuilder<>(null, new StrictBufferConfigImpl(MAX_VALUE, 2L, SHUT_DOWN))
+            is(new FinalResultsSuppressionBuilder<>(null, new StrictBufferConfigImpl(MAX_VALUE, 2L, SHUT_DOWN, Collections.emptyMap()))
             )
         );
 
@@ -122,13 +124,13 @@ public void finalEventsShouldAcceptStrictBuffersAndSetBounds() {
 
         assertThat(
             untilWindowCloses(maxRecords(2L).shutDownWhenFull()).withName("name"),
-            is(new FinalResultsSuppressionBuilder<>("name", new StrictBufferConfigImpl(2L, MAX_VALUE, SHUT_DOWN))
+            is(new FinalResultsSuppressionBuilder<>("name", new StrictBufferConfigImpl(2L, MAX_VALUE, SHUT_DOWN, Collections.emptyMap()))
             )
         );
 
         assertThat(
             untilWindowCloses(maxBytes(2L).shutDownWhenFull()).withName("name"),
-            is(new FinalResultsSuppressionBuilder<>("name", new StrictBufferConfigImpl(MAX_VALUE, 2L, SHUT_DOWN))
+            is(new FinalResultsSuppressionBuilder<>("name", new StrictBufferConfigImpl(MAX_VALUE, 2L, SHUT_DOWN, Collections.emptyMap()))
             )
         );
     }

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Suppressed.scala

@@ -17,7 +17,6 @@
 package org.apache.kafka.streams.scala.kstream
 
 import java.time.Duration
-
 import org.apache.kafka.streams.kstream.{Windowed, Suppressed => SupressedJ}
 import org.apache.kafka.streams.kstream.Suppressed.{
   EagerBufferConfig,
@@ -30,6 +29,7 @@ import org.apache.kafka.streams.kstream.internals.suppress.{
   StrictBufferConfigImpl,
   SuppressedInternal
 }
+import java.util.Collections
 
 /**
  * Duplicates the static factory methods inside the Java interface [[org.apache.kafka.streams.kstream.Suppressed]].
@@ -93,7 +93,7 @@ object Suppressed {
      * @see [[org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxRecords]]
      */
     def maxRecords(recordLimit: Long): EagerBufferConfig =
-      new EagerBufferConfigImpl(recordLimit, Long.MaxValue)
+      new EagerBufferConfigImpl(recordLimit, Long.MaxValue, Collections.emptyMap())
 
     /**
      * Create a size-constrained buffer in terms of the maximum number of bytes it will use.
@@ -103,7 +103,7 @@ object Suppressed {
      * @see [[org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxBytes]]
      */
     def maxBytes(byteLimit: Long): EagerBufferConfig =
-      new EagerBufferConfigImpl(Long.MaxValue, byteLimit)
+      new EagerBufferConfigImpl(Long.MaxValue, byteLimit, Collections.emptyMap())
 
     /**
      * Create a buffer unconstrained by size (either keys or bytes).

streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/SuppressedTest.scala

@@ -22,6 +22,7 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 
 import java.time.Duration
+import java.util.Collections
 
 @deprecated(message = "org.apache.kafka.streams.scala.kstream.Suppressed has been deprecated", since = "2.5")
 class SuppressedTest {
@@ -44,20 +45,34 @@ class SuppressedTest {
 
   @Test
   def testProduceCorrectBufferConfigWithMaxRecords(): Unit = {
-    assertEquals(BufferConfig.maxRecords(4), new EagerBufferConfigImpl(4, Long.MaxValue))
-    assertEquals(BufferConfig.maxRecords(4).withMaxBytes(5), new EagerBufferConfigImpl(4, 5))
+    assertEquals(
+      BufferConfig.maxRecords(4),
+      new EagerBufferConfigImpl(4, Long.MaxValue, Collections.emptyMap())
+    )
+    assertEquals(
+      BufferConfig.maxRecords(4).withMaxBytes(5),
+      new EagerBufferConfigImpl(4, 5, Collections.emptyMap())
+    )
   }
 
   @Test
   def testProduceCorrectBufferConfigWithMaxBytes(): Unit = {
-    assertEquals(BufferConfig.maxBytes(4), new EagerBufferConfigImpl(Long.MaxValue, 4))
-    assertEquals(BufferConfig.maxBytes(4).withMaxRecords(5), new EagerBufferConfigImpl(5, 4))
+    assertEquals(
+      BufferConfig.maxBytes(4),
+      new EagerBufferConfigImpl(Long.MaxValue, 4, Collections.emptyMap())
+    )
+    assertEquals(
+      BufferConfig.maxBytes(4).withMaxRecords(5),
+      new EagerBufferConfigImpl(5, 4, Collections.emptyMap())
+    )
   }
 
   @Test
   def testProduceCorrectBufferConfigWithUnbounded(): Unit =
-    assertEquals(BufferConfig.unbounded(),
-                 new StrictBufferConfigImpl(Long.MaxValue, Long.MaxValue, BufferFullStrategy.SHUT_DOWN))
+    assertEquals(
+      BufferConfig.unbounded(),
+      new StrictBufferConfigImpl(Long.MaxValue, Long.MaxValue, BufferFullStrategy.SHUT_DOWN, Collections.emptyMap())
+    )
 
   @Test
   def testSupportLongChainsOfFactoryMethods(): Unit = {
@@ -68,8 +83,9 @@ class SuppressedTest {
       .withMaxBytes(4L)
       .withMaxRecords(5L)
       .withMaxBytes(6L)
-    assertEquals(new EagerBufferConfigImpl(5L, 6L), bc1)
-    assertEquals(new StrictBufferConfigImpl(5L, 6L, BufferFullStrategy.SHUT_DOWN), bc1.shutDownWhenFull())
+    assertEquals(new EagerBufferConfigImpl(5L, 6L, Collections.emptyMap()), bc1)
+    assertEquals(new StrictBufferConfigImpl(5L, 6L, BufferFullStrategy.SHUT_DOWN, Collections.emptyMap()),
+                 bc1.shutDownWhenFull())
 
     val bc2 = BufferConfig
       .maxBytes(4)
@@ -79,7 +95,7 @@ class SuppressedTest {
       .withMaxBytes(7)
       .withMaxRecords(8)
 
-    assertEquals(new StrictBufferConfigImpl(8L, 7L, BufferFullStrategy.SHUT_DOWN), bc2)
+    assertEquals(new StrictBufferConfigImpl(8L, 7L, BufferFullStrategy.SHUT_DOWN, Collections.emptyMap()), bc2)
     assertEquals(BufferConfig.unbounded(), bc2.withNoBound())
 
     val bc3 = BufferConfig
@@ -88,6 +104,6 @@ class SuppressedTest {
       .emitEarlyWhenFull()
       .withMaxRecords(11L)
 
-    assertEquals(new EagerBufferConfigImpl(11L, 10L), bc3)
+    assertEquals(new EagerBufferConfigImpl(11L, 10L, Collections.emptyMap()), bc3)
   }
 }