Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 merged PR #20847: URL: https://github.com/apache/kafka/pull/20847 -- 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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on PR #20847: URL: https://github.com/apache/kafka/pull/20847#issuecomment-3642120034 @squah-confluent @dajac I will merge this patch. Because it is not slated for 4.2.0, we have plenty of time to resolve any feedback in subsequent work -- 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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on code in PR #20847: URL: https://github.com/apache/kafka/pull/20847#discussion_r2606470217 ## docs/upgrade.html: ## @@ -19,6 +19,19 @@
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
squah-confluent commented on code in PR #20847: URL: https://github.com/apache/kafka/pull/20847#discussion_r2606434326 ## docs/upgrade.html: ## @@ -19,6 +19,19 @@
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on code in PR #20847: URL: https://github.com/apache/kafka/pull/20847#discussion_r2606397937 ## docs/upgrade.html: ## @@ -19,6 +19,19 @@
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on code in PR #20847:
URL: https://github.com/apache/kafka/pull/20847#discussion_r2605033129
##
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##
@@ -99,7 +101,9 @@ object DynamicBrokerConfig {
SocketServer.ReconfigurableConfigs ++
DynamicProducerStateManagerConfig ++
DynamicRemoteLogConfig.ReconfigurableConfigs ++
-Set(AbstractConfig.CONFIG_PROVIDERS_CONFIG)
+Set(AbstractConfig.CONFIG_PROVIDERS_CONFIG) ++
+GroupCoordinatorConfig.RECONFIGURABLE_CONFIGS.asScala ++
Review Comment:
Do we have integration tests for it?
--
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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on code in PR #20847:
URL: https://github.com/apache/kafka/pull/20847#discussion_r2605024705
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##
@@ -312,6 +324,8 @@ public class GroupCoordinatorConfig {
.define(OFFSETS_TOPIC_PARTITIONS_CONFIG, INT,
OFFSETS_TOPIC_PARTITIONS_DEFAULT, atLeast(1), HIGH,
OFFSETS_TOPIC_PARTITIONS_DOC)
.define(OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG, INT,
OFFSETS_TOPIC_SEGMENT_BYTES_DEFAULT, atLeast(1), HIGH,
OFFSETS_TOPIC_SEGMENT_BYTES_DOC)
.define(OFFSETS_TOPIC_COMPRESSION_CODEC_CONFIG, INT, (int)
OFFSETS_TOPIC_COMPRESSION_CODEC_DEFAULT.id, HIGH,
OFFSETS_TOPIC_COMPRESSION_CODEC_DOC)
+.define(CACHED_BUFFER_MAX_BYTES_CONFIG, INT,
CACHED_BUFFER_MAX_BYTES_DEFAULT, atLeast(512 * 1024), MEDIUM,
Review Comment:
The minimum size is equal to `INITIAL_BUFFER_SIZE`. Could you please add a
comment explaining this connection?
--
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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on code in PR #20847:
URL: https://github.com/apache/kafka/pull/20847#discussion_r2605016022
##
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##
@@ -772,13 +780,20 @@ private void freeCurrentBatch() {
// Cancel the linger timeout.
currentBatch.lingerTimeoutTask.ifPresent(TimerTask::cancel);
-// Release the buffer only if it is not larger than the
maxBatchSize.
-int maxBatchSize = partitionWriter.config(tp).maxMessageSize();
+// Release the buffer only if it is not larger than the
cachedBufferMaxBytes.
+int cachedBufferMaxBytes = cachedBufferMaxBytesSupplier.get();
-if (currentBatch.builder.buffer().capacity() <= maxBatchSize) {
+if (currentBatch.builder.buffer().capacity() <=
cachedBufferMaxBytes) {
bufferSupplier.release(currentBatch.builder.buffer());
-} else if (currentBatch.buffer.capacity() <= maxBatchSize) {
+cachedBufferSize.set(currentBatch.builder.buffer().capacity());
+} else if (currentBatch.buffer.capacity() <= cachedBufferMaxBytes)
{
bufferSupplier.release(currentBatch.buffer);
+cachedBufferSize.set(currentBatch.buffer.capacity());
+// If the builder expands the buffer beyond the
cachedBufferMaxBytes, that should also increment the discard counter.
Review Comment:
`increment` -> `increase`
--
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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on code in PR #20847: URL: https://github.com/apache/kafka/pull/20847#discussion_r2605012840 ## docs/ops.html: ## @@ -1937,6 +1937,16 @@
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on PR #20847: URL: https://github.com/apache/kafka/pull/20847#issuecomment-3631162909 > I'm in favor of removing them from the GroupCoordinator interface methods as a separate PR. I can't foresee us using them again any time soon. Thanks for confirming. @mingyen066 please open a patch once this PR is shipped. -- 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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
squah-confluent commented on PR #20847: URL: https://github.com/apache/kafka/pull/20847#issuecomment-3630481901 @chia7712 I'm in favor of removing them from the `GroupCoordinator` interface methods. I can't foresee us using them again any time soon. -- 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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
chia7712 commented on PR #20847: URL: https://github.com/apache/kafka/pull/20847#issuecomment-3622035499 The `BufferSupplier` parameter in `GroupCoordinator` and `ShareCoordinator` is unused. If its original design was to handle records, it is time to remove it from the method signature. @squah-confluent @dajac WDYT? -- 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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
squah-confluent commented on code in PR #20847:
URL: https://github.com/apache/kafka/pull/20847#discussion_r2594677110
##
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java:
##
@@ -69,6 +69,16 @@ public class CoordinatorRuntimeMetricsImpl implements
CoordinatorRuntimeMetrics
*/
public static final String BATCH_FLUSH_TIME_METRIC_NAME =
"batch-flush-time-ms";
+/**
+ * The cached buffer size metric name.
Review Comment:
nit:
```suggestion
* The buffer cache size metric name.
```
##
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java:
##
@@ -92,6 +102,17 @@ public class CoordinatorRuntimeMetricsImpl implements
CoordinatorRuntimeMetrics
*/
private final MetricName eventQueueSize;
+/**
+ * Metric to count the size of the cached buffer.
Review Comment:
nit:
```suggestion
* Metric to count the size of the cached buffers.
```
##
coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImplTest.java:
##
@@ -59,6 +61,8 @@ private static Set expectedMetricNames(Metrics
metrics) {
kafkaMetricName(metrics, NUM_PARTITIONS_METRIC_NAME, "state",
"loading"),
kafkaMetricName(metrics, NUM_PARTITIONS_METRIC_NAME, "state",
"active"),
kafkaMetricName(metrics, NUM_PARTITIONS_METRIC_NAME, "state",
"failed"),
+kafkaMetricName(metrics, BATCH_BUFFER_CACHE_SIZE_METRIC_NAME),
+kafkaMetricName(metrics,
BATCH_BUFFER_CACHE_DISCARD_COUNT_METRIC_NAME),
Review Comment:
nit: Can we move these down after the `batch-flush-rate`? so that they're
together with the batch- metrics.
##
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java:
##
@@ -156,6 +177,17 @@ public CoordinatorRuntimeMetricsImpl(Metrics metrics,
String metricsGroup) {
this.eventQueueSize = kafkaMetricName("event-queue-size", "The event
accumulator queue size.");
+this.bufferCacheSize = kafkaMetricName(
+BATCH_BUFFER_CACHE_SIZE_METRIC_NAME,
+"The current total size in bytes of the append buffers being held
in the coordinator's cache."
+);
+
+this.bufferCacheDiscardCount = kafkaMetricName(
+BATCH_BUFFER_CACHE_DISCARD_COUNT_METRIC_NAME,
+"The count of over-sized append buffers that were discarded
instead of being cached upon release."
+);
+
+metrics.addMetric(bufferCacheDiscardCount, (Gauge) (config, now)
-> bufferCacheDiscardCounter.get());
Review Comment:
nit: Could we move this after `metrics.addMetric(numPartitionsFailed` to
match the order of the field definitions?
##
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java:
##
@@ -69,6 +69,16 @@ public class CoordinatorRuntimeMetricsImpl implements
CoordinatorRuntimeMetrics
*/
public static final String BATCH_FLUSH_TIME_METRIC_NAME =
"batch-flush-time-ms";
+/**
+ * The cached buffer size metric name.
+ */
+public static final String BATCH_BUFFER_CACHE_SIZE_METRIC_NAME =
"batch-buffer-cache-size-bytes";
+
+/**
+ * The buffer skip cache count metric name.
Review Comment:
nit:
```suggestion
* The buffer cache discard count metric name.
```
##
docs/upgrade.html:
##
@@ -19,16 +19,26 @@
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
squah-confluent commented on PR #20847: URL: https://github.com/apache/kafka/pull/20847#issuecomment-3619809150 > Furthermore, since the coordinator usually uses "bytes" to represent size limits, we should align the suffix to reflect that standard 😄 I missed that, that's a good idea! -- 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]
Re: [PR] KAFKA-19519: Introduce group/share.coordinator.cached.buffer.max.bytes config [kafka]
DL1231 commented on PR #20847: URL: https://github.com/apache/kafka/pull/20847#issuecomment-3612350311 @chia7712 @squah-confluent, thanks for the review, I have updated the PR, please take another look. And I'll update the KIP later. -- 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]
