WindowBytesStoreSupplier 的正确配置

Question

什么是 JoinWindows.of(Duration.ofMinutes(5)) 的正确 WindowBytesStoreSupplier 配置值？

这个

fun mangelJoinStoreSupplier(): WindowBytesStoreSupplier = Stores.inMemoryWindowStore(MangelJoinStore.NAME, Duration.ofMillis(86700000), Duration.ofMinutes(5), true)

结果：

org.apache.kafka.streams.errors.StreamsException: Window settings mismatch. WindowBytesStoreSupplier settings InMemoryWindowBytesStoreSupplier{name='mangel-join-store', retentionPeriod=86700000, windowSize=300000, retainDuplicates=true} must match JoinWindows settings JoinWindows{beforeMs=300000, afterMs=300000, graceMs=85800000} for the window size and retention period

Answer 1

为了

JoinWindows.of(Duration.ofMinutes(5)).grace(Duration.ZERO)

是

Stores.inMemoryWindowStore(MangelJoinStore.NAME, Duration.ofMinutes(10), Duration.ofMinutes(10), true)

正确的供应商