You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/04/01 16:55:56 UTC
(kafka) branch 4.3 updated (b368e39a4d7 -> 4852527b5cc)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a change to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
from b368e39a4d7 KAFKA-20366: Add syntax highlighting across docs, excluding security and streams (#21887)
new b89f34af14c MINOR: simplify window-byte-store-supplier (#21829)
new 4852527b5cc MINOR: deprecate internal config for NamedTopologies (#21921)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/kafka/streams/StreamsConfig.java | 1 +
.../internals/InternalTopologyBuilder.java | 2 +
.../processor/internals/ProcessorContextUtils.java | 1 +
.../org/apache/kafka/streams/state/Stores.java | 50 ++++------------
...IndexedTimeOrderedWindowBytesStoreSupplier.java | 69 ++++++++++++++--------
.../internals/RocksDbWindowBytesStoreSupplier.java | 26 +++-----
.../internals/TimeOrderedCachingWindowStore.java | 1 +
7 files changed, 68 insertions(+), 82 deletions(-)
(kafka) 02/02: MINOR: deprecate internal config for NamedTopologies (#21921)
Posted by mj...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 4852527b5cc1a27c0ea1d4028a113719fd21adc1
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Wed Apr 1 09:50:16 2026 -0700
MINOR: deprecate internal config for NamedTopologies (#21921)
NamedTopologies got deprecated via KAFKA-17805. This PR deprecates the
internal "topic prefix" config, which is used by NamedTopologies.
Reviewers: Christo Lolov <lo...@amazon.com>, TengYao Chi
<fr...@apache.org>
---
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java | 1 +
.../kafka/streams/processor/internals/InternalTopologyBuilder.java | 2 ++
.../apache/kafka/streams/processor/internals/ProcessorContextUtils.java | 1 +
.../kafka/streams/state/internals/TimeOrderedCachingWindowStore.java | 1 +
4 files changed, 5 insertions(+)
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index d28818e47e3..2d860cd7905 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -1362,6 +1362,7 @@ public class StreamsConfig extends AbstractConfig {
+ ".vector.enabled__";
// Private API used to control the prefix of the auto created topics
+ @Deprecated
public static final String TOPIC_PREFIX_ALTERNATIVE = "__internal.override.topic.prefix__";
// Private API to enable processing threads (i.e. polling is decoupled from processing)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
index 86f1ebbe7a6..efdbf7d5974 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
@@ -1122,6 +1122,7 @@ public class InternalTopologyBuilder {
}
}
+ @SuppressWarnings("deprecation")
private void buildProcessorNode(final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap,
final Map<String, StateStore> stateStoreMap,
final ProcessorNodeFactory<?, ?, ?, ?> factory,
@@ -1456,6 +1457,7 @@ public class InternalTopologyBuilder {
return decorateTopic(topic);
}
+ @SuppressWarnings("deprecation")
private String decorateTopic(final String topic) {
if (applicationId == null) {
throw new TopologyException("there are internal topics and "
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java
index 0515f8718aa..aade59a4f7d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java
@@ -52,6 +52,7 @@ public final class ProcessorContextUtils {
return ProcessorStateManager.storeChangelogTopic(prefix, storeName, context.taskId().topologyName());
}
+ @Deprecated
public static String topicNamePrefix(final Map<String, Object> configs, final String applicationId) {
if (configs == null) {
return applicationId;
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
index aea172ea9cc..4e0fd29f09d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
@@ -113,6 +113,7 @@ public class TimeOrderedCachingWindowStore
return null;
}
+ @SuppressWarnings("deprecation")
@Override
public void init(final StateStoreContext stateStoreContext, final StateStore root) {
final String prefix = StreamsConfig.InternalConfig.getString(
(kafka) 01/02: MINOR: simplify window-byte-store-supplier (#21829)
Posted by mj...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit b89f34af14c616076571d690e755b1272de02522
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Fri Mar 27 16:26:50 2026 -0700
MINOR: simplify window-byte-store-supplier (#21829)
This PR removes an unnecessary "duplicated" constructor for
RocksDbWindowBytesStoreSupplier, and unifies the caller stack to re-use
more shared code.
Also make internal contractors non-public for
RocksDbIndexedTimeOrderedWindowBytesStoreSupplier.
Reviewers: Lan Ding <is...@163.com>, Alieh Saeedi
<as...@confluent.io>, TengYao Chi <fr...@apache.org>
---
.../org/apache/kafka/streams/state/Stores.java | 50 ++++------------
...IndexedTimeOrderedWindowBytesStoreSupplier.java | 69 ++++++++++++++--------
.../internals/RocksDbWindowBytesStoreSupplier.java | 26 +++-----
3 files changed, 63 insertions(+), 82 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
index db39656b83b..c541eca32a0 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
@@ -299,7 +299,7 @@ public final class Stores {
final Duration retentionPeriod,
final Duration windowSize,
final boolean retainDuplicates) throws IllegalArgumentException {
- return persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates, false);
+ return persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates, RocksDbWindowBytesStoreSupplier.WindowStoreTypes.DEFAULT_WINDOW_STORE);
}
/**
@@ -331,7 +331,7 @@ public final class Stores {
final Duration retentionPeriod,
final Duration windowSize,
final boolean retainDuplicates) throws IllegalArgumentException {
- return persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates, true);
+ return persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates, RocksDbWindowBytesStoreSupplier.WindowStoreTypes.TIMESTAMPED_WINDOW_STORE);
}
/**
@@ -348,28 +348,14 @@ public final class Stores {
final Duration retentionPeriod,
final Duration windowSize,
final boolean retainDuplicates) throws IllegalArgumentException {
- Objects.requireNonNull(name, "name cannot be null");
- final String rpMsgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod");
- final long retentionMs = validateMillisecondDuration(retentionPeriod, rpMsgPrefix);
- final String wsMsgPrefix = prepareMillisCheckFailMsgPrefix(windowSize, "windowSize");
- final long windowSizeMs = validateMillisecondDuration(windowSize, wsMsgPrefix);
-
- final long defaultSegmentInterval = Math.max(retentionMs / 2, 60_000L);
-
- return new RocksDbWindowBytesStoreSupplier(
- name,
- retentionMs,
- defaultSegmentInterval,
- windowSizeMs,
- retainDuplicates,
- RocksDbWindowBytesStoreSupplier.WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS);
+ return persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates, RocksDbWindowBytesStoreSupplier.WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS);
}
private static WindowBytesStoreSupplier persistentWindowStore(final String name,
final Duration retentionPeriod,
final Duration windowSize,
final boolean retainDuplicates,
- final boolean timestampedStore) {
+ final RocksDbWindowBytesStoreSupplier.WindowStoreTypes storeType) {
Objects.requireNonNull(name, "name cannot be null");
final String rpMsgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod");
final long retentionMs = validateMillisecondDuration(retentionPeriod, rpMsgPrefix);
@@ -378,26 +364,13 @@ public final class Stores {
final long defaultSegmentInterval = Math.max(retentionMs / 2, 60_000L);
- return persistentWindowStore(name, retentionMs, windowSizeMs, retainDuplicates, defaultSegmentInterval, timestampedStore);
- }
-
- private static WindowBytesStoreSupplier persistentWindowStore(final String name,
- final long retentionPeriod,
- final long windowSize,
- final boolean retainDuplicates,
- final long segmentInterval,
- final boolean timestampedStore) {
- Objects.requireNonNull(name, "name cannot be null");
- if (retentionPeriod < 0L) {
+ if (retentionMs < 0L) {
throw new IllegalArgumentException("retentionPeriod cannot be negative");
}
- if (windowSize < 0L) {
+ if (windowSizeMs < 0L) {
throw new IllegalArgumentException("windowSize cannot be negative");
}
- if (segmentInterval < 1L) {
- throw new IllegalArgumentException("segmentInterval cannot be zero or negative");
- }
- if (windowSize > retentionPeriod) {
+ if (windowSizeMs > retentionMs) {
throw new IllegalArgumentException("The retention period of the window store "
+ name + " must be no smaller than its window size. Got size=["
+ windowSize + "], retention=[" + retentionPeriod + "]");
@@ -405,11 +378,12 @@ public final class Stores {
return new RocksDbWindowBytesStoreSupplier(
name,
- retentionPeriod,
- segmentInterval,
- windowSize,
+ retentionMs,
+ defaultSegmentInterval,
+ windowSizeMs,
retainDuplicates,
- timestampedStore);
+ storeType
+ );
}
/**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbIndexedTimeOrderedWindowBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbIndexedTimeOrderedWindowBytesStoreSupplier.java
index 968821eb5c4..4e372242f8d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbIndexedTimeOrderedWindowBytesStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbIndexedTimeOrderedWindowBytesStoreSupplier.java
@@ -40,12 +40,14 @@ public class RocksDbIndexedTimeOrderedWindowBytesStoreSupplier implements Window
private final boolean retainDuplicates;
private final WindowStoreTypes windowStoreType;
- public static RocksDbIndexedTimeOrderedWindowBytesStoreSupplier create(final String name,
- final Duration retentionPeriod,
- final Duration windowSize,
- final boolean retainDuplicates,
- final boolean hasIndex,
- final boolean withHeaders) {
+ public static RocksDbIndexedTimeOrderedWindowBytesStoreSupplier create(
+ final String name,
+ final Duration retentionPeriod,
+ final Duration windowSize,
+ final boolean retainDuplicates,
+ final boolean hasIndex,
+ final boolean withHeaders
+ ) {
Objects.requireNonNull(name, "name cannot be null");
final String rpMsgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod");
final long retentionMs = validateMillisecondDuration(retentionPeriod, rpMsgPrefix);
@@ -60,28 +62,41 @@ public class RocksDbIndexedTimeOrderedWindowBytesStoreSupplier implements Window
if (windowSizeMs < 0L) {
throw new IllegalArgumentException("windowSize cannot be negative");
}
- if (defaultSegmentInterval < 1L) {
- throw new IllegalArgumentException("segmentInterval cannot be zero or negative");
- }
if (windowSizeMs > retentionMs) {
throw new IllegalArgumentException("The retention period of the window store "
+ name + " must be no smaller than its window size. Got size=["
+ windowSizeMs + "], retention=[" + retentionMs + "]");
}
- return new RocksDbIndexedTimeOrderedWindowBytesStoreSupplier(name, retentionMs,
- defaultSegmentInterval, windowSizeMs, retainDuplicates, hasIndex, withHeaders);
+ return new RocksDbIndexedTimeOrderedWindowBytesStoreSupplier(
+ name,
+ retentionMs,
+ defaultSegmentInterval,
+ windowSizeMs,
+ retainDuplicates,
+ hasIndex,
+ withHeaders
+ );
}
- public RocksDbIndexedTimeOrderedWindowBytesStoreSupplier(final String name,
- final long retentionPeriod,
- final long segmentInterval,
- final long windowSize,
- final boolean retainDuplicates,
- final boolean withIndex,
- final boolean withHeaders) {
- this(name, retentionPeriod, segmentInterval, windowSize, retainDuplicates,
- determineStoreType(withIndex, withHeaders));
+ // for testing only
+ RocksDbIndexedTimeOrderedWindowBytesStoreSupplier(
+ final String name,
+ final long retentionPeriod,
+ final long segmentInterval,
+ final long windowSize,
+ final boolean retainDuplicates,
+ final boolean withIndex,
+ final boolean withHeaders
+ ) {
+ this(
+ name,
+ retentionPeriod,
+ segmentInterval,
+ windowSize,
+ retainDuplicates,
+ determineStoreType(withIndex, withHeaders)
+ );
}
private static WindowStoreTypes determineStoreType(final boolean withIndex, final boolean withHeaders) {
@@ -94,12 +109,14 @@ public class RocksDbIndexedTimeOrderedWindowBytesStoreSupplier implements Window
}
}
- public RocksDbIndexedTimeOrderedWindowBytesStoreSupplier(final String name,
- final long retentionPeriod,
- final long segmentInterval,
- final long windowSize,
- final boolean retainDuplicates,
- final WindowStoreTypes windowStoreType) {
+ private RocksDbIndexedTimeOrderedWindowBytesStoreSupplier(
+ final String name,
+ final long retentionPeriod,
+ final long segmentInterval,
+ final long windowSize,
+ final boolean retainDuplicates,
+ final WindowStoreTypes windowStoreType
+ ) {
this.name = name;
this.retentionPeriod = retentionPeriod;
this.segmentInterval = segmentInterval;
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java
index 14213b659e5..0d958d82822 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java
@@ -34,24 +34,14 @@ public class RocksDbWindowBytesStoreSupplier implements WindowBytesStoreSupplier
private final boolean retainDuplicates;
private final WindowStoreTypes windowStoreType;
- public RocksDbWindowBytesStoreSupplier(final String name,
- final long retentionPeriod,
- final long segmentInterval,
- final long windowSize,
- final boolean retainDuplicates,
- final boolean returnTimestampedStore) {
- this(name, retentionPeriod, segmentInterval, windowSize, retainDuplicates,
- returnTimestampedStore
- ? WindowStoreTypes.TIMESTAMPED_WINDOW_STORE
- : WindowStoreTypes.DEFAULT_WINDOW_STORE);
- }
-
- public RocksDbWindowBytesStoreSupplier(final String name,
- final long retentionPeriod,
- final long segmentInterval,
- final long windowSize,
- final boolean retainDuplicates,
- final WindowStoreTypes windowStoreType) {
+ public RocksDbWindowBytesStoreSupplier(
+ final String name,
+ final long retentionPeriod,
+ final long segmentInterval,
+ final long windowSize,
+ final boolean retainDuplicates,
+ final WindowStoreTypes windowStoreType
+ ) {
this.name = name;
this.retentionPeriod = retentionPeriod;
this.segmentInterval = segmentInterval;