bbejeck commented on code in PR #22597:
URL: https://github.com/apache/kafka/pull/22597#discussion_r3438931578
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/StateStoreMetrics.java:
##########
@@ -563,4 +567,20 @@ public static void addNumKeysGauge(final String taskId,
numKeysGauge
);
}
+
+ public static void addUncommittedBytesGauge(final String taskId,
+ final String storeType,
Review Comment:
nit: formatting align `final` keywords with first one
##########
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. " +
Review Comment:
nit: also divided across a global thread if it exists
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java:
##########
@@ -301,6 +312,13 @@ public void run() {
}
stateConsumer.pollAndUpdate();
+ final long uncommittedLimit = maxUncommittedBytes;
Review Comment:
I get that but I'm wondering if we used `AtomicLong` would offer any benefit
over the current approach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
