.

LGouellec · Apr 13, 2024 · 8692e62 · 8692e62
1 parent 7b10cbc
commit 8692e62
Show file tree

Hide file tree

Showing 6 changed files with 9 additions and 9 deletions.
diff --git a/core/State/Logging/ChangeLoggingTimestampedWindowBytesStore.cs b/core/State/Logging/ChangeLoggingTimestampedWindowBytesStore.cs
@@ -6,10 +6,9 @@ namespace Streamiz.Kafka.Net.State.Logging
     internal class ChangeLoggingTimestampedWindowBytesStore :
         ChangeLoggingWindowBytesStore
     {
-        public ChangeLoggingTimestampedWindowBytesStore(IWindowStore<Bytes, byte[]> wrapped) 
-            : base(wrapped)
-        {
-        }
+        public ChangeLoggingTimestampedWindowBytesStore(IWindowStore<Bytes, byte[]> wrapped, bool retainDuplicates) 
+            : base(wrapped, retainDuplicates)
+        { }
 
         protected override void Publish(Bytes key, byte[] valueAndTs)
         {

diff --git a/core/State/Logging/ChangeLoggingWindowBytesStore.cs b/core/State/Logging/ChangeLoggingWindowBytesStore.cs
@@ -13,9 +13,10 @@ internal class ChangeLoggingWindowBytesStore :
         private int seqnum = 0;
         private readonly bool retainDuplicates;
 
-        public ChangeLoggingWindowBytesStore(IWindowStore<Bytes, byte[]> wrapped) 
+        public ChangeLoggingWindowBytesStore(IWindowStore<Bytes, byte[]> wrapped, bool retainDuplicates) 
             : base(wrapped)
         {
+            this.retainDuplicates = retainDuplicates;
         }
 
         protected virtual void Publish(Bytes key, byte[] value)

diff --git a/core/State/TimestampedWindowStoreBuilder.cs b/core/State/TimestampedWindowStoreBuilder.cs
@@ -58,7 +58,7 @@ private IWindowStore<Bytes, byte[]> WrapLogging(IWindowStore<Bytes, byte[]> inne
             if (!LoggingEnabled)
                 return inner;
 
-            return new ChangeLoggingTimestampedWindowBytesStore(inner);
+            return new ChangeLoggingTimestampedWindowBytesStore(inner, supplier.RetainDuplicates);
         }
     }
 }
diff --git a/core/State/WindowStoreBuilder.cs b/core/State/WindowStoreBuilder.cs
@@ -59,7 +59,7 @@ private IWindowStore<Bytes, byte[]> WrapLogging(IWindowStore<Bytes, byte[]> inne
             if (!LoggingEnabled)
                 return inner;
 
-            return new ChangeLoggingWindowBytesStore(inner);
+            return new ChangeLoggingWindowBytesStore(inner, supplier.RetainDuplicates);
         }
     }
 }
diff --git a/test/Streamiz.Kafka.Net.Tests/Stores/ChangeLoggingTimestampedWindowBytesStoreTests.cs b/test/Streamiz.Kafka.Net.Tests/Stores/ChangeLoggingTimestampedWindowBytesStoreTests.cs
@@ -74,7 +74,7 @@ public void Begin()
             context.UseRecordCollector(recordCollector);
 
             var inmemorystore = new InMemoryWindowStore("test-store", TimeSpan.FromDays(1), TimeSpan.FromSeconds(1).Milliseconds, false);
-            store = new ChangeLoggingTimestampedWindowBytesStore(inmemorystore);
+            store = new ChangeLoggingTimestampedWindowBytesStore(inmemorystore, false);
             store.Init(context, store);
         }
 

diff --git a/test/Streamiz.Kafka.Net.Tests/Stores/ChangeLoggingWindowBytesStoreTests.cs b/test/Streamiz.Kafka.Net.Tests/Stores/ChangeLoggingWindowBytesStoreTests.cs
@@ -70,7 +70,7 @@ public void Begin()
             context.UseRecordCollector(recordCollector);
 
             var inmemorystore = new InMemoryWindowStore("test-store", TimeSpan.FromDays(1), TimeSpan.FromSeconds(1).Milliseconds, false);
-            store = new ChangeLoggingWindowBytesStore(inmemorystore);
+            store = new ChangeLoggingWindowBytesStore(inmemorystore, false);
             store.Init(context, store);
         }