This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 6110d3e1ba4 KAFKA-20491: Add uncommitted bytes limit (#22597)
6110d3e1ba4 is described below
commit 6110d3e1ba47fe92ad776eae6b5d64cdcc19ddbf
Author: Nick Telford <[email protected]>
AuthorDate: Fri Jun 19 16:03:40 2026 +0100
KAFKA-20491: Add uncommitted bytes limit (#22597)
Introduces the `statestore.uncommitted.max.bytes` config, which triggers
an early commit when the total uncommitted bytes across all
transactional state stores on a thread exceeds the configured limit,
regardless of `commit.interval.ms`. This bounds the memory consumed by
pending write buffers under high write throughput.
The limit is divided equally across StreamThreads and the
GlobalStreamThread. Each thread independently enforces its share:
StreamThreads trigger an early task commit via `maybeCommit()`; the
GlobalStreamThread flushes at the end of each `pollAndUpdate()` cycle.
`StateStore` gains a default `approximateNumUncommittedBytes()` method
(returning 0); adapter and wrapper classes delegate to the inner store.
Metered key-value stores register an uncommitted-bytes gauge. Segmented
and versioned stores aggregate across their underlying segments.
The per-thread limit is recomputed whenever stream threads are
dynamically added or removed via
`addStreamThread()`/`removeStreamThread()`, mirroring the existing
thread-cache resize logic. The per-thread limit field is `volatile` so
running threads pick up the new value on their next commit check without
synchronisation.
This is part of
[KIP-892](https://cwiki.apache.org/confluence/display/KAFKA/KIP-892%3A+Transactional+Semantics+for+State+Stores).
Reviewers: Bill Bejeck <[email protected]>
---
.../org/apache/kafka/streams/KafkaStreams.java | 31 ++-
.../org/apache/kafka/streams/StreamsConfig.java | 15 ++
.../apache/kafka/streams/processor/StateStore.java | 9 +
.../processor/internals/GlobalStateMaintainer.java | 2 +
.../processor/internals/GlobalStateManager.java | 2 +
.../internals/GlobalStateManagerImpl.java | 11 +
.../processor/internals/GlobalStateUpdateTask.java | 5 +
.../processor/internals/GlobalStreamThread.java | 22 ++
.../processor/internals/ProcessorStateManager.java | 8 +
.../streams/processor/internals/ReadOnlyTask.java | 5 +
.../streams/processor/internals/StreamTask.java | 5 +
.../streams/processor/internals/StreamThread.java | 21 +-
.../kafka/streams/processor/internals/Task.java | 4 +
.../streams/processor/internals/TaskManager.java | 8 +
.../AbstractRocksDBSegmentedBytesStore.java | 9 +
.../state/internals/KeyValueStoreWrapper.java | 5 +
...ValueToTimestampedKeyValueByteStoreAdapter.java | 5 +
.../state/internals/LogicalKeyValueSegments.java | 4 +
.../state/internals/MeteredKeyValueStore.java | 4 +
.../internals/PlainToHeadersStoreAdapter.java | 5 +
.../PlainToHeadersWindowStoreAdapter.java | 5 +
.../state/internals/RocksDBVersionedStore.java | 6 +
.../internals/SessionToHeadersStoreAdapter.java | 5 +
.../TimestampedToHeadersStoreAdapter.java | 5 +
.../TimestampedToHeadersWindowStoreAdapter.java | 5 +
.../VersionedKeyValueToBytesStoreAdapter.java | 5 +
.../WindowToTimestampedWindowByteStoreAdapter.java | 5 +
.../streams/state/internals/WrappedStateStore.java | 5 +
.../state/internals/metrics/StateStoreMetrics.java | 20 ++
.../org/apache/kafka/streams/KafkaStreamsTest.java | 1 +
.../internals/GlobalStateManagerImplTest.java | 28 +++
.../processor/internals/GlobalStateTaskTest.java | 6 +
.../internals/GlobalStreamThreadTest.java | 26 +++
.../processor/internals/StateConsumerTest.java | 5 +
.../processor/internals/StreamThreadTest.java | 236 +++++++++++++++++++--
.../apache/kafka/test/GlobalStateManagerStub.java | 6 +
36 files changed, 522 insertions(+), 27 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index 851fced582c..e5f1ee47c54 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -1012,6 +1012,7 @@ public class KafkaStreams implements AutoCloseable {
totalCacheSize = totalCacheSize(applicationConfigs);
final int numStreamThreads =
topologyMetadata.numStreamThreads(applicationConfigs);
final long cacheSizePerThread = cacheSizePerThread(numStreamThreads);
+ final long maxUncommittedBytesPerThread =
maxUncommittedBytesPerThread(numStreamThreads);
GlobalStreamThread.State globalThreadState = null;
if (hasGlobalTopology) {
@@ -1022,6 +1023,7 @@ public class KafkaStreams implements AutoCloseable {
clientSupplier.getGlobalConsumer(applicationConfigs.getGlobalConsumerConfigs(clientId)),
stateDirectory,
cacheSizePerThread,
+ maxUncommittedBytesPerThread,
streamsMetrics,
time,
globalThreadId,
@@ -1043,7 +1045,7 @@ public class KafkaStreams implements AutoCloseable {
globalStateStoreProvider,
applicationConfigs::defaultInteractiveQueryIsolationLevel);
for (int i = 1; i <= numStreamThreads; i++) {
- createAndAddStreamThread(cacheSizePerThread, i);
+ createAndAddStreamThread(cacheSizePerThread,
maxUncommittedBytesPerThread, i);
}
stateDirCleaner = setupStateDirCleaner();
@@ -1066,7 +1068,9 @@ public class KafkaStreams implements AutoCloseable {
}
}
- private StreamThread createAndAddStreamThread(final long
cacheSizePerThread, final int threadIdx) {
+ private StreamThread createAndAddStreamThread(final long
cacheSizePerThread,
+ final long
maxUncommittedBytesPerThread,
+ final int threadIdx) {
final StreamThread streamThread = StreamThread.create(
topologyMetadata,
applicationConfigs,
@@ -1078,6 +1082,7 @@ public class KafkaStreams implements AutoCloseable {
time,
streamsMetadataState,
cacheSizePerThread,
+ maxUncommittedBytesPerThread,
stateDirectory,
delegatingStateRestoreListener,
delegatingStandbyUpdateListener,
@@ -1124,12 +1129,14 @@ public class KafkaStreams implements AutoCloseable {
final int threadIdx = nextThreadIndex();
final int numLiveThreads = numLiveStreamThreads();
final long cacheSizePerThread =
cacheSizePerThread(numLiveThreads + 1);
+ final long maxUncommittedBytesPerThread =
maxUncommittedBytesPerThread(numLiveThreads + 1);
log.info("Adding StreamThread-{}, there will now be {} live
threads and the new cache size per thread is {}",
threadIdx, numLiveThreads + 1, cacheSizePerThread);
resizeThreadCache(cacheSizePerThread);
+ resizeMaxUncommittedBytes(maxUncommittedBytesPerThread);
// Creating thread should hold the lock in order to avoid
duplicate thread index.
// If the duplicate index happen, the metadata of thread may
be duplicate too.
- streamThread = createAndAddStreamThread(cacheSizePerThread,
threadIdx);
+ streamThread = createAndAddStreamThread(cacheSizePerThread,
maxUncommittedBytesPerThread, threadIdx);
}
synchronized (stateLock) {
@@ -1143,6 +1150,7 @@ public class KafkaStreams implements AutoCloseable {
final long cacheSizePerThread =
cacheSizePerThread(numLiveStreamThreads());
log.info("Resizing thread cache due to terminating added
thread, new cache size per thread is {}", cacheSizePerThread);
resizeThreadCache(cacheSizePerThread);
+
resizeMaxUncommittedBytes(maxUncommittedBytesPerThread(numLiveStreamThreads()));
return Optional.empty();
}
}
@@ -1223,6 +1231,7 @@ public class KafkaStreams implements AutoCloseable {
final long cacheSizePerThread =
cacheSizePerThread(numLiveStreamThreads());
log.info("Resizing thread cache due to thread removal,
new cache size per thread is {}", cacheSizePerThread);
resizeThreadCache(cacheSizePerThread);
+
resizeMaxUncommittedBytes(maxUncommittedBytesPerThread(numLiveStreamThreads()));
final long remainingTimeMs = timeoutMs -
(time.milliseconds() - startMs);
if (remainingTimeMs <= 0) {
throw new TimeoutException("Thread " +
streamThread.getName() + " did not stop in the allotted time");
@@ -1324,6 +1333,15 @@ public class KafkaStreams implements AutoCloseable {
return totalCacheSize / (numStreamThreads +
(topologyMetadata.hasGlobalTopology() ? 1 : 0));
}
+ private long maxUncommittedBytesPerThread(final int numStreamThreads) {
+ final long totalMax =
applicationConfigs.getLong(StreamsConfig.STATESTORE_UNCOMMITTED_MAX_BYTES_CONFIG);
+ if (totalMax <= 0) {
+ return -1;
+ }
+ final int divisor = Math.max(numStreamThreads, 0) +
(topologyMetadata.hasGlobalTopology() ? 1 : 0);
+ return divisor == 0 ? totalMax : totalMax / divisor;
+ }
+
private void resizeThreadCache(final long cacheSizePerThread) {
processStreamThread(thread -> thread.resizeCache(cacheSizePerThread));
if (globalStreamThread != null) {
@@ -1331,6 +1349,13 @@ public class KafkaStreams implements AutoCloseable {
}
}
+ private void resizeMaxUncommittedBytes(final long
maxUncommittedBytesPerThread) {
+ processStreamThread(thread ->
thread.resizeMaxUncommittedBytes(maxUncommittedBytesPerThread));
+ if (globalStreamThread != null) {
+
globalStreamThread.resizeMaxUncommittedBytes(maxUncommittedBytesPerThread);
+ }
+ }
+
private ScheduledExecutorService setupStateDirCleaner() {
return Executors.newSingleThreadScheduledExecutor(r -> {
final Thread thread = new Thread(r, clientId + "-CleanupThread");
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 b227c058a81..d9e1569fe36 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -808,6 +808,15 @@ public class StreamsConfig extends AbstractConfig {
@Deprecated
public static final String STATESTORE_CACHE_MAX_BYTES_DOC = "Maximum
number of memory bytes to be used for statestore cache across all threads";
+ /** {@code statestore.uncommitted.max.bytes} */
+ public static final String STATESTORE_UNCOMMITTED_MAX_BYTES_CONFIG =
"statestore.uncommitted.max.bytes";
+ private static final String STATESTORE_UNCOMMITTED_MAX_BYTES_DOC =
+ "The maximum number of uncommitted bytes across all transactional
state stores on this " +
+ "application instance before an early commit is triggered, regardless
of commit.interval.ms. " +
+ "The limit is divided equally across the configured number of stream
threads, and the global state thread," +
+ "if the topology has any global stores. Set to -1 to disable. Default
is 67108864 (64 MB).";
+ private static final long STATESTORE_UNCOMMITTED_MAX_BYTES_DEFAULT =
67_108_864L;
+
/** {@code task.assignor.class} */
@SuppressWarnings("WeakerAccess")
public static final String TASK_ASSIGNOR_CLASS_CONFIG =
"task.assignor.class";
@@ -959,6 +968,12 @@ public class StreamsConfig extends AbstractConfig {
atLeast(0),
Importance.MEDIUM,
STATESTORE_CACHE_MAX_BYTES_DOC)
+ .define(STATESTORE_UNCOMMITTED_MAX_BYTES_CONFIG,
+ Type.LONG,
+ STATESTORE_UNCOMMITTED_MAX_BYTES_DEFAULT,
+ atLeast(-1),
+ Importance.LOW,
+ STATESTORE_UNCOMMITTED_MAX_BYTES_DOC)
.define(CLIENT_ID_CONFIG,
Type.STRING,
"",
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
b/streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
index 6f91222f64b..19112419ddc 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
@@ -215,6 +215,15 @@ public interface StateStore {
return QueryResult.forUnknownQueryType(query, this);
}
+ /**
+ * Returns an approximation of the number of uncommitted bytes currently
buffered in this store's
+ * transaction buffer, or 0 if the store is not transactional. This is
used to trigger early
+ * commits when the buffer grows too large.
+ */
+ default long approximateNumUncommittedBytes() {
+ return 0;
+ }
+
/**
* Returns the position the state store is at with respect to the input
topic/partitions
*/
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java
index 06afb6fde4f..8c4c993459c 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateMaintainer.java
@@ -36,4 +36,6 @@ interface GlobalStateMaintainer {
void update(ConsumerRecord<byte[], byte[]> record);
void maybeCheckpoint();
+
+ long approximateNumUncommittedBytes();
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
index f470254142e..3baa7d99025 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManager.java
@@ -29,4 +29,6 @@ public interface GlobalStateManager extends StateManager {
* @throws StreamsException if the store's change log does not contain the
partition
*/
Set<String> initialize();
+
+ long approximateNumUncommittedBytes();
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
index 8e7347e6fa4..fe7f7028904 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java
@@ -570,6 +570,17 @@ public class GlobalStateManagerImpl implements
GlobalStateManager {
return currentDeadlineMs;
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ long total = 0;
+ for (final Optional<StateStore> entry : globalStores.values()) {
+ if (entry.isPresent()) {
+ total += entry.get().approximateNumUncommittedBytes();
+ }
+ }
+ return total;
+ }
+
@Override
public void commit() {
log.debug("Committing all global globalStores registered in the state
manager");
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
index 37178598451..45d21d619e9 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
@@ -168,6 +168,11 @@ public class GlobalStateUpdateTask implements
GlobalStateMaintainer {
}
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return stateMgr.approximateNumUncommittedBytes();
+ }
+
@Override
public void maybeCheckpoint() {
final long now = time.milliseconds();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
index bede888525a..4b5bc646e0d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
@@ -75,6 +75,7 @@ public class GlobalStreamThread extends Thread {
private volatile long fetchDeadlineClientInstanceId = -1;
private volatile KafkaFutureImpl<Uuid> clientInstanceIdFuture = new
KafkaFutureImpl<>();
private final CountDownLatch initializationLatch = new CountDownLatch(1);
+ private volatile long maxUncommittedBytes;
/**
* The states that the global stream thread can be in
@@ -202,6 +203,7 @@ public class GlobalStreamThread extends Thread {
final Consumer<byte[], byte[]> globalConsumer,
final StateDirectory stateDirectory,
final long cacheSizeBytes,
+ final long maxUncommittedBytes,
final StreamsMetricsImpl streamsMetrics,
final Time time,
final String threadClientId,
@@ -221,6 +223,7 @@ public class GlobalStreamThread extends Thread {
this.stateRestoreListener = stateRestoreListener;
this.streamsUncaughtExceptionHandler = streamsUncaughtExceptionHandler;
this.cacheSize = new AtomicLong(-1L);
+ this.maxUncommittedBytes = maxUncommittedBytes;
}
static class StateConsumer {
@@ -259,6 +262,14 @@ public class GlobalStreamThread extends Thread {
stateMaintainer.maybeCheckpoint();
}
+ void flushState() {
+ stateMaintainer.flushState();
+ }
+
+ long approximateNumUncommittedBytes() {
+ return stateMaintainer.approximateNumUncommittedBytes();
+ }
+
public void close(final boolean wipeStateStore) throws IOException {
try {
globalConsumer.close();
@@ -301,6 +312,13 @@ public class GlobalStreamThread extends Thread {
}
stateConsumer.pollAndUpdate();
+ final long uncommittedLimit = maxUncommittedBytes;
+ if (uncommittedLimit > 0
+ && stateConsumer.approximateNumUncommittedBytes() >
uncommittedLimit) {
+ log.debug("Committing global state: uncommitted bytes
exceeded {}", uncommittedLimit);
+ stateConsumer.flushState();
+ }
+
if (fetchDeadlineClientInstanceId != -1) {
if (fetchDeadlineClientInstanceId >= time.milliseconds()) {
try {
@@ -372,6 +390,10 @@ public class GlobalStreamThread extends Thread {
this.cacheSize.set(cacheSize);
}
+ public void resizeMaxUncommittedBytes(final long maxUncommittedBytes) {
+ this.maxUncommittedBytes = maxUncommittedBytes;
+ }
+
private StateConsumer initialize() {
StateConsumer stateConsumer = null;
try {
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
index d6a5a2fea3c..6d620a56a52 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
@@ -629,6 +629,14 @@ public class ProcessorStateManager implements StateManager
{
}
}
+ long approximateNumUncommittedBytes() {
+ long total = 0;
+ for (final StateStoreMetadata metadata : stores.values()) {
+ total += metadata.stateStore.approximateNumUncommittedBytes();
+ }
+ return total;
+ }
+
/**
* {@link StateStore#close() Close} all stores (even in case of failure).
* Log all exceptions and re-throw the first exception that occurred at
the end.
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
index e81fabc704f..eaaf9db01e1 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
@@ -246,4 +246,9 @@ public class ReadOnlyTask implements Task {
public ProcessorStateManager stateManager() {
throw new UnsupportedOperationException("This task is read-only");
}
+
+ @Override
+ public long approximateNumUncommittedBytes() {
+ throw new UnsupportedOperationException("This task is read-only");
+ }
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 3b65e677674..2ab631fcfaa 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -1351,6 +1351,11 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
return sb.toString();
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return stateMgr.approximateNumUncommittedBytes();
+ }
+
@Override
public boolean commitNeeded() {
// we need to do an extra check if the flag was false, that
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index b265c3fc276..c5c7e3b87cb 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -330,6 +330,7 @@ public class StreamThread extends Thread implements
ProcessingThread {
private final Sensor commitRatioSensor;
private final Sensor failedStreamThreadSensor;
+ private volatile long maxUncommittedBytesPerThread;
private final long logSummaryIntervalMs; // the count summary log output
time interval
private long lastLogSummaryMs = -1L;
private long totalRecordsProcessedSinceLastSummary = 0L;
@@ -406,6 +407,7 @@ public class StreamThread extends Thread implements
ProcessingThread {
final Time time,
final StreamsMetadataState
streamsMetadataState,
final long cacheSizeBytes,
+ final long maxUncommittedBytesPerThread,
final StateDirectory stateDirectory,
final StateRestoreListener
userStateRestoreListener,
final StandbyUpdateListener
userStandbyUpdateListener,
@@ -542,7 +544,8 @@ public class StreamThread extends Thread implements
ProcessingThread {
cache::resize,
mainConsumerSetup.streamsRebalanceData,
streamsMetadataState,
- metricsReporter
+ metricsReporter,
+ maxUncommittedBytesPerThread
);
return streamThread.updateThreadMetadata(adminClientId(clientId));
@@ -790,7 +793,8 @@ public class StreamThread extends Thread implements
ProcessingThread {
final java.util.function.Consumer<Long> cacheResizer,
final Optional<StreamsRebalanceData>
streamsRebalanceData,
final StreamsMetadataState streamsMetadataState,
- final StreamsThreadMetricsDelegatingReporter
metricsReporter
+ final StreamsThreadMetricsDelegatingReporter
metricsReporter,
+ final long maxUncommittedBytesPerThread
) {
super(threadId);
this.stateLock = new Object();
@@ -885,6 +889,7 @@ public class StreamThread extends Thread implements
ProcessingThread {
this.eosEnabled = eosEnabled(config);
this.processingThreadsEnabled =
InternalConfig.processingThreadsEnabled(config.originals());
this.logSummaryIntervalMs =
config.getLong(StreamsConfig.LOG_SUMMARY_INTERVAL_MS_CONFIG);
+ this.maxUncommittedBytesPerThread = maxUncommittedBytesPerThread;
this.streamsRebalanceData = streamsRebalanceData;
this.streamsMetadataState = streamsMetadataState;
@@ -1816,7 +1821,7 @@ public class StreamThread extends Thread implements
ProcessingThread {
// visible for testing
int maybeCommit() {
final int committed;
- if (now - lastCommitMs > commitTimeMs) {
+ if (now - lastCommitMs > commitTimeMs ||
shouldCommitDueToUncommittedBytes()) {
if (log.isDebugEnabled()) {
log.debug("Committing all active tasks {} and standby tasks {}
since {}ms has elapsed (commit interval is {}ms)",
taskManager.activeRunningTaskIds(),
taskManager.standbyTaskIds(), now - lastCommitMs, commitTimeMs);
@@ -1849,6 +1854,16 @@ public class StreamThread extends Thread implements
ProcessingThread {
return committed;
}
+ private boolean shouldCommitDueToUncommittedBytes() {
+ final long limit = maxUncommittedBytesPerThread;
+ if (limit <= 0) return false;
+ return taskManager.totalUncommittedBytes() > limit;
+ }
+
+ public void resizeMaxUncommittedBytes(final long
maxUncommittedBytesPerThread) {
+ this.maxUncommittedBytesPerThread = maxUncommittedBytesPerThread;
+ }
+
/**
* Compute the latency based on the current marked timestamp, and update
the marked timestamp
* with the current system timestamp.
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
index e2404badef3..eb7f45042e7 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
@@ -238,6 +238,10 @@ public interface Task {
boolean commitNeeded();
+ default long approximateNumUncommittedBytes() {
+ return 0;
+ }
+
default boolean commitRequested() {
return false;
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
index a25e0e32f88..0e2723d07af 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
@@ -1690,6 +1690,14 @@ public class TaskManager {
return tasks.allInitializedTasksPerId();
}
+ long totalUncommittedBytes() {
+ long total = 0;
+ for (final Task task : allRunningTasks().values()) {
+ total += task.approximateNumUncommittedBytes();
+ }
+ return total;
+ }
+
Set<Task> readOnlyAllTasks() {
// not bothering with an unmodifiable map, since the tasks themselves
are mutable, but
// if any outside code modifies the map or the tasks, it would be a
severe transgression.
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
index c11e276a19e..9357583ec53 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
@@ -334,6 +334,15 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
return segments.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ long total = 0;
+ for (final S segment : segments.allSegments(true)) {
+ total += segment.approximateNumUncommittedBytes();
+ }
+ return total;
+ }
+
@Override
public void close() {
open = false;
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapper.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapper.java
index 62d939562ca..836a609c184 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapper.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapper.java
@@ -151,6 +151,11 @@ public class KeyValueStoreWrapper<K, V> implements
StateStore {
store.commit(changelogOffsets);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
index 3612776fbeb..5c5ededb8b4 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
@@ -117,6 +117,11 @@ public class KeyValueToTimestampedKeyValueByteStoreAdapter
implements KeyValueSt
return store.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegments.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegments.java
index 9fbdf762c8d..90f15a33daa 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegments.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegments.java
@@ -55,6 +55,10 @@ public class LogicalKeyValueSegments extends
AbstractSegments<LogicalKeyValueSeg
this.physicalStore = new RocksDBStore(name, parentDir,
metricsRecorder, false);
}
+ RocksDBStore physicalStore() {
+ return physicalStore;
+ }
+
@Override
protected LogicalKeyValueSegment createSegment(final long segmentId, final
String segmentName) {
if (segmentId < 0) {
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
index 327ba3ff69a..ca8f12ac743 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
@@ -190,6 +190,10 @@ public class MeteredKeyValueStore<K, V>
}
);
}
+ StateStoreMetrics.addUncommittedBytesGauge(
+ taskId.toString(), metricsScope, name(), streamsMetrics,
+ (config, now) -> wrapped().approximateNumUncommittedBytes()
+ );
}
@Override
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersStoreAdapter.java
index 17415dbe207..4fa41db8ece 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersStoreAdapter.java
@@ -121,6 +121,11 @@ public class PlainToHeadersStoreAdapter implements
KeyValueStore<Bytes, byte[]>
return store.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java
index a463dbc97f2..218182878e8 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java
@@ -232,6 +232,11 @@ public class PlainToHeadersWindowStoreAdapter implements
WindowStore<Bytes, byte
return store.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
index e9c705972ee..c4c26d047f5 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
@@ -320,6 +320,12 @@ public class RocksDBVersionedStore implements
VersionedKeyValueStore<Bytes, byte
return segmentStores.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ // All logical segments share one physical store — its buffer holds
all uncommitted bytes
+ return segmentStores.physicalStore().approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
open = false;
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java
index db753e52bbe..91c79234325 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java
@@ -167,6 +167,11 @@ public class SessionToHeadersStoreAdapter implements
SessionStore<Bytes, byte[]>
return store.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java
index facef5f0a06..918893b77d5 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java
@@ -123,6 +123,11 @@ public class TimestampedToHeadersStoreAdapter implements
KeyValueStore<Bytes, by
return store.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
index cde8ac607d8..889533f5534 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
@@ -179,6 +179,11 @@ public class TimestampedToHeadersWindowStoreAdapter
implements WindowStore<Bytes
return store.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
index c02cec5747a..ee422bd837e 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
@@ -110,6 +110,11 @@ public class VersionedKeyValueToBytesStoreAdapter
implements VersionedBytesStore
return inner.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return inner.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
inner.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
index 5a7f05c1b2b..eb037c77534 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
@@ -178,6 +178,11 @@ public class WindowToTimestampedWindowByteStoreAdapter
implements WindowStore<By
return store.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return store.approximateNumUncommittedBytes();
+ }
+
@Override
public void close() {
store.close();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
index 73786170a27..6032feb362a 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
@@ -174,6 +174,11 @@ public abstract class WrappedStateStore<S extends
StateStore, K, V> implements S
return result;
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return wrapped.approximateNumUncommittedBytes();
+ }
+
@Override
public Position getPosition() {
return wrapped.getPosition();
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java
index 551ee1d0983..8a235a5db5f 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java
@@ -169,6 +169,10 @@ public class StateStoreMetrics {
private static final String NUM_KEYS_DESCRIPTION =
"The current number of keys in the in-memory state store";
+ private static final String UNCOMMITTED_BYTES = "uncommitted-bytes";
+ private static final String UNCOMMITTED_BYTES_DESCRIPTION =
+ "The approximate number of uncommitted bytes in the transactional
state store buffer";
+
private static final String OLDEST_ITERATOR_OPEN_SINCE_MS =
"oldest-iterator-open-since-ms";
private static final String OLDEST_ITERATOR_OPEN_SINCE_MS_DESCRIPTION =
"The UNIX timestamp the oldest still open iterator was created, in
milliseconds";
@@ -563,4 +567,20 @@ public class StateStoreMetrics {
numKeysGauge
);
}
+
+ public static void addUncommittedBytesGauge(final String taskId,
+ final String storeType,
+ final String storeName,
+ final StreamsMetricsImpl
streamsMetrics,
+ final Gauge<Long>
uncommittedBytesGauge) {
+ streamsMetrics.addStoreLevelMutableMetric(
+ taskId,
+ storeType,
+ storeName,
+ UNCOMMITTED_BYTES,
+ UNCOMMITTED_BYTES_DESCRIPTION,
+ RecordingLevel.INFO,
+ uncommittedBytesGauge
+ );
+ }
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
index 4ee975805dc..e1d8af5d318 100644
--- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
@@ -255,6 +255,7 @@ public class KafkaStreamsTest {
any(Time.class),
any(StreamsMetadataState.class),
anyLong(),
+ anyLong(),
any(StateDirectory.class),
any(StateRestoreListener.class),
any(StandbyUpdateListener.class),
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
index 4759bdc0508..c5eefac3bca 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
@@ -394,6 +394,34 @@ public class GlobalStateManagerImplTest {
}
+ @Test
+ public void
shouldReportZeroApproximateNumUncommittedBytesBeforeStoresAreRegistered() {
+ initializeConsumer(0, 0, t1, t2, t3, t4, t5);
+ stateManager.initialize();
+ assertEquals(0L, stateManager.approximateNumUncommittedBytes());
+ }
+
+ @Test
+ public void
shouldAggregateApproximateNumUncommittedBytesAcrossRegisteredStores() {
+ initializeConsumer(0, 0, t1, t2, t3, t4, t5);
+ stateManager.initialize();
+ initializeConsumer(1, 0, t1);
+ stateManager.registerStore(new NoOpReadOnlyStore<>(store1.name()) {
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return 100L;
+ }
+ }, stateRestoreCallback, null);
+ initializeConsumer(1, 0, t2);
+ stateManager.registerStore(new NoOpReadOnlyStore<>(store2.name()) {
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return 250L;
+ }
+ }, stateRestoreCallback, null);
+ assertEquals(350L, stateManager.approximateNumUncommittedBytes());
+ }
+
@Test
public void shouldCommitStateStores() {
initializeConsumer(0, 0, t1, t2, t3, t4, t5);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java
index d5f22be3a2b..c082f9d6bc7 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java
@@ -246,6 +246,12 @@ public class GlobalStateTaskTest {
}
+ @Test
+ public void shouldDelegateApproximateNumUncommittedBytesToStateManager() {
+ stateMgr.approximateNumUncommittedBytes = 4242L;
+ assertEquals(4242L, globalStateTask.approximateNumUncommittedBytes());
+ }
+
@Test
public void shouldCommitStateManagerWithOffsets() {
final Map<TopicPartition, Long> expectedOffsets = new HashMap<>();
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
index 7c0216cd07f..8a77af5a2ea 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
@@ -132,6 +132,7 @@ public class GlobalStreamThreadTest {
mockConsumer,
new StateDirectory(config, time, true, false),
0,
+ -1L,
new StreamsMetricsImpl(new Metrics(), "test-client", time),
time,
"clientId",
@@ -169,6 +170,7 @@ public class GlobalStreamThreadTest {
mockConsumer,
new StateDirectory(config, time, true, false),
0,
+ -1L,
new StreamsMetricsImpl(new Metrics(), "test-client", time),
time,
"clientId",
@@ -188,6 +190,29 @@ public class GlobalStreamThreadTest {
assertFalse(globalStreamThread.stillRunning());
}
+ @Test
+ public void shouldAllowResizingMaxUncommittedBytesBeforeStart() {
+ // Invoking the resize should be safe before the thread has started.
+ globalStreamThread.resizeMaxUncommittedBytes(4096L);
+ }
+
+ @Test
+ public void shouldAllowResizingMaxUncommittedBytesWhileRunning() throws
Exception {
+ initializeConsumer();
+ startAndSwallowError();
+
+ TestUtils.waitForCondition(
+ () -> globalStreamThread.state() == RUNNING,
+ 10 * 1000,
+ "Thread never started.");
+
+ globalStreamThread.resizeMaxUncommittedBytes(4096L);
+ assertTrue(globalStreamThread.stillRunning());
+
+ globalStreamThread.shutdown();
+ globalStreamThread.join();
+ }
+
@Test
public void shouldBeRunningAfterSuccessfulStart() throws Exception {
initializeConsumer();
@@ -418,6 +443,7 @@ public class GlobalStreamThreadTest {
consumer,
new StateDirectory(config, time, true, false),
0,
+ -1L,
new StreamsMetricsImpl(new Metrics(), "test-client", time),
time,
"clientId",
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
index 72c0fc38d8c..a779678098e 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
@@ -145,6 +145,11 @@ public class StateConsumerTest {
public void maybeCheckpoint() {
flushState();
}
+
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return 0;
+ }
}
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index fb8bb26c0be..95292c15899 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -328,6 +328,7 @@ public class StreamThreadTest {
time,
streamsMetadataState,
0,
+ -1L,
stateDirectory,
new MockStateRestoreListener(),
new MockStandbyUpdateListener(),
@@ -764,6 +765,7 @@ public class StreamThreadTest {
mockTime,
streamsMetadataState,
0,
+ -1L,
stateDirectory,
new MockStateRestoreListener(),
new MockStandbyUpdateListener(),
@@ -826,6 +828,7 @@ public class StreamThreadTest {
mockTime,
streamsMetadataState,
0,
+ -1L,
stateDirectory,
new MockStateRestoreListener(),
new MockStandbyUpdateListener(),
@@ -1100,6 +1103,166 @@ public class StreamThreadTest {
assertTrue(committed.get());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldCommitWhenUncommittedBytesExceedLimit(final boolean
processingThreadsEnabled) {
+ final long commitInterval = Duration.ofMinutes(1).toMillis();
+ final long maxUncommittedBytesPerThread = 1024L;
+
+ final Properties props = configProps(false, processingThreadsEnabled);
+ props.setProperty(StreamsConfig.STATE_DIR_CONFIG, stateDir);
+ props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,
Long.toString(commitInterval));
+
+ final StreamsConfig config = new StreamsConfig(props);
+ final ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ when(consumer.groupMetadata()).thenReturn(consumerGroupMetadata);
+
when(consumerGroupMetadata.groupInstanceId()).thenReturn(Optional.empty());
+
+ final AtomicBoolean committed = new AtomicBoolean(false);
+ final AtomicLong uncommittedBytes = new AtomicLong(0L);
+ final TopologyMetadata topologyMetadata = new
TopologyMetadata(internalTopologyBuilder, config);
+ final TaskManager taskManager = new TaskManager(
+ null, null, null, null, null, null,
+ new Tasks(new LogContext()),
+ topologyMetadata,
+ null, null, null, null
+ ) {
+ @Override
+ int commit(final Collection<? extends Task> tasksToCommit) {
+ committed.set(true);
+ return 1;
+ }
+
+ @Override
+ long totalUncommittedBytes() {
+ return uncommittedBytes.get();
+ }
+ };
+ topologyMetadata.buildAndRewriteTopology();
+ thread = buildStreamThread(consumer, taskManager, config,
topologyMetadata, maxUncommittedBytesPerThread);
+
+ // prime lastCommitMs so the interval-based trigger does not fire
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ committed.set(false);
+
+ // below the limit: no commit
+ uncommittedBytes.set(maxUncommittedBytesPerThread);
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ assertFalse(committed.get());
+
+ // exceeding the limit triggers an early commit regardless of commit
interval
+ uncommittedBytes.set(maxUncommittedBytesPerThread + 1);
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ assertTrue(committed.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldNotCommitOnUncommittedBytesWhenLimitIsDisabled(final
boolean processingThreadsEnabled) {
+ final long commitInterval = Duration.ofMinutes(1).toMillis();
+
+ final Properties props = configProps(false, processingThreadsEnabled);
+ props.setProperty(StreamsConfig.STATE_DIR_CONFIG, stateDir);
+ props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,
Long.toString(commitInterval));
+
+ final StreamsConfig config = new StreamsConfig(props);
+ final ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ when(consumer.groupMetadata()).thenReturn(consumerGroupMetadata);
+
when(consumerGroupMetadata.groupInstanceId()).thenReturn(Optional.empty());
+
+ final AtomicBoolean committed = new AtomicBoolean(false);
+ final TopologyMetadata topologyMetadata = new
TopologyMetadata(internalTopologyBuilder, config);
+ final TaskManager taskManager = new TaskManager(
+ null, null, null, null, null, null,
+ new Tasks(new LogContext()),
+ topologyMetadata,
+ null, null, null, null
+ ) {
+ @Override
+ int commit(final Collection<? extends Task> tasksToCommit) {
+ committed.set(true);
+ return 1;
+ }
+
+ @Override
+ long totalUncommittedBytes() {
+ return Long.MAX_VALUE;
+ }
+ };
+ topologyMetadata.buildAndRewriteTopology();
+ thread = buildStreamThread(consumer, taskManager, config,
topologyMetadata, -1L);
+
+ // prime lastCommitMs so the interval-based trigger does not fire
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ committed.set(false);
+
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ assertFalse(committed.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldResizeMaxUncommittedBytes(final boolean
processingThreadsEnabled) {
+ final long commitInterval = Duration.ofMinutes(1).toMillis();
+ final long initialLimit = 1024L;
+ final long resizedLimit = 4096L;
+
+ final Properties props = configProps(false, processingThreadsEnabled);
+ props.setProperty(StreamsConfig.STATE_DIR_CONFIG, stateDir);
+ props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,
Long.toString(commitInterval));
+
+ final StreamsConfig config = new StreamsConfig(props);
+ final ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ when(consumer.groupMetadata()).thenReturn(consumerGroupMetadata);
+
when(consumerGroupMetadata.groupInstanceId()).thenReturn(Optional.empty());
+
+ final AtomicBoolean committed = new AtomicBoolean(false);
+ final AtomicLong uncommittedBytes = new AtomicLong(0L);
+ final TopologyMetadata topologyMetadata = new
TopologyMetadata(internalTopologyBuilder, config);
+ final TaskManager taskManager = new TaskManager(
+ null, null, null, null, null, null,
+ new Tasks(new LogContext()),
+ topologyMetadata,
+ null, null, null, null
+ ) {
+ @Override
+ int commit(final Collection<? extends Task> tasksToCommit) {
+ committed.set(true);
+ return 1;
+ }
+
+ @Override
+ long totalUncommittedBytes() {
+ return uncommittedBytes.get();
+ }
+ };
+ topologyMetadata.buildAndRewriteTopology();
+ thread = buildStreamThread(consumer, taskManager, config,
topologyMetadata, initialLimit);
+
+ // prime lastCommitMs so the interval-based trigger does not fire
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ committed.set(false);
+
+ // usage between initial and resized limit: commits under initial
limit, but
+ // should no longer commit once the limit is raised above the usage
+ uncommittedBytes.set(initialLimit + 1);
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ assertTrue(committed.get());
+
+ thread.resizeMaxUncommittedBytes(resizedLimit);
+ committed.set(false);
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+ assertFalse(committed.get());
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void shouldPurgeAfterPurgeInterval(final boolean
processingThreadsEnabled) {
@@ -1453,7 +1616,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
final StreamsException thrown = assertThrows(StreamsException.class,
thread::run);
@@ -1549,7 +1713,7 @@ public class StreamThreadTest {
PROCESS_ID, CLIENT_ID, new LogContext(""),
null, new AtomicLong(Long.MAX_VALUE), new LinkedList<>(),
null, HANDLER, null,
- Optional.of(streamsRebalanceData), null, null
+ Optional.of(streamsRebalanceData), null, null, -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.shutdown(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.DEFAULT);
@@ -1582,7 +1746,7 @@ public class StreamThreadTest {
PROCESS_ID, CLIENT_ID, new LogContext(""),
null, new AtomicLong(Long.MAX_VALUE), new LinkedList<>(),
null, HANDLER, null,
- Optional.of(streamsRebalanceData), null, null
+ Optional.of(streamsRebalanceData), null, null, -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.shutdown(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP);
@@ -2008,6 +2172,7 @@ public class StreamThreadTest {
mockTime,
streamsMetadataState,
0,
+ -1L,
stateDirectory,
new MockStateRestoreListener(),
new MockStandbyUpdateListener(),
@@ -2571,7 +2736,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
) {
@Override
void runOnceWithProcessingThreads() {
@@ -2633,7 +2799,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
) {
@Override
void runOnceWithProcessingThreads() {
@@ -2703,7 +2870,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
) {
@Override
void runOnceWithProcessingThreads() {
@@ -2770,7 +2938,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
) {
@Override
void runOnceWithProcessingThreads() {
@@ -2842,7 +3011,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
null,
- null
+ null,
+ -1L
) {
@Override
void runOnceWithProcessingThreads() {
@@ -2909,7 +3079,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
) {
@Override
void runOnceWithProcessingThreads() {
@@ -3142,7 +3313,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
);
final MetricName testMetricName = new MetricName("test_metric", "",
"", new HashMap<>());
final Metric testMetric = new KafkaMetric(
@@ -3202,7 +3374,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
) {
@Override
void runOnceWithProcessingThreads() {
@@ -3581,6 +3754,7 @@ public class StreamThreadTest {
mockTime,
streamsMetadataState,
0,
+ -1L,
stateDirectory,
new MockStateRestoreListener(),
new MockStandbyUpdateListener(),
@@ -3639,6 +3813,7 @@ public class StreamThreadTest {
mockTime,
streamsMetadataState,
0,
+ -1L,
stateDirectory,
new MockStateRestoreListener(),
new MockStandbyUpdateListener(),
@@ -3704,6 +3879,7 @@ public class StreamThreadTest {
mockTime,
streamsMetadataState,
0,
+ -1L,
stateDirectory,
new MockStateRestoreListener(),
new MockStandbyUpdateListener(),
@@ -3790,7 +3966,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.setState(State.STARTING);
@@ -3849,7 +4026,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.sendShutdownRequest();
@@ -3896,7 +4074,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.sendShutdownRequest();
@@ -3951,7 +4130,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.setState(State.STARTING);
@@ -4022,7 +4202,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.setState(State.STARTING);
@@ -4084,7 +4265,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.setState(State.STARTING);
@@ -4146,7 +4328,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.setState(State.STARTING);
@@ -4217,7 +4400,8 @@ public class StreamThreadTest {
null,
Optional.of(streamsRebalanceData),
streamsMetadataState,
- null
+ null,
+ -1L
).updateThreadMetadata(adminClientId(CLIENT_ID));
thread.setState(State.STARTING);
@@ -4342,7 +4526,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ -1L
);
}
@@ -4418,6 +4603,14 @@ public class StreamThreadTest {
final TaskManager taskManager,
final StreamsConfig config,
final TopologyMetadata
topologyMetadata) {
+ return buildStreamThread(consumer, taskManager, config,
topologyMetadata, -1L);
+ }
+
+ private StreamThread buildStreamThread(final Consumer<byte[], byte[]>
consumer,
+ final TaskManager taskManager,
+ final StreamsConfig config,
+ final TopologyMetadata
topologyMetadata,
+ final long
maxUncommittedBytesPerThread) {
final StreamsMetricsImpl streamsMetrics =
new StreamsMetricsImpl(metrics, CLIENT_ID, mockTime);
@@ -4444,7 +4637,8 @@ public class StreamThreadTest {
null,
Optional.empty(),
null,
- null
+ null,
+ maxUncommittedBytesPerThread
);
}
diff --git
a/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
b/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
index 1237deacf9b..354762737a0 100644
--- a/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
+++ b/streams/src/test/java/org/apache/kafka/test/GlobalStateManagerStub.java
@@ -36,6 +36,7 @@ public class GlobalStateManagerStub implements
GlobalStateManager {
public boolean initialized;
public boolean closed;
public boolean committed;
+ public long approximateNumUncommittedBytes;
public GlobalStateManagerStub(final Set<String> storeNames,
final Map<TopicPartition, Long> offsets,
@@ -69,6 +70,11 @@ public class GlobalStateManagerStub implements
GlobalStateManager {
committed = true;
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ return approximateNumUncommittedBytes;
+ }
+
@Override
public void close() {
closed = true;
