from:"GitBox"

[GitHub] [kafka] fvaleri commented on pull request #13136: KAFKA-14582: Move JmxTool to tools

2023-01-20 Thread GitBox



fvaleri commented on PR #13136:
URL: https://github.com/apache/kafka/pull/13136#issuecomment-1398707514

   Output example:
   
   ```sh
   $ bin/kafka-run-class.sh org.apache.kafka.tools.JmxCommand --jmx-url 
service:jmx:rmi:///jndi/rmi://:/jmxrmi --object-name 
kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec --attributes 
FifteenMinuteRate,FiveMinuteRate --date-format "MMdd-hh:mm:ss" 
--reporting-interval 1000 --report-format tsv
   Trying to connect to JMX url: service:jmx:rmi:///jndi/rmi://:/jmxrmi
   time20230120-06:23:14
   kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate 
   0.0
   kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate
0.0
   time 20230120-06:23:15
   kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate 
0.0
   kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate
0.0
   time 20230120-06:23:16
   kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate 
0.0
   kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate
0.0
   ^C
   ```


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] kamalcph commented on pull request #13060: KAFKA-14559: Fix JMX tool to handle the object names with wild cards and optional attributes

2023-01-20 Thread GitBox



kamalcph commented on PR #13060:
URL: https://github.com/apache/kafka/pull/13060#issuecomment-1398673057

   @ijuma @showuon 
   Please take a look when you get chance.


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] kamalcph commented on pull request #13059: MINOR: KafkaConfig should not expose internal config when queried for non-internal values

2023-01-20 Thread GitBox



kamalcph commented on PR #13059:
URL: https://github.com/apache/kafka/pull/13059#issuecomment-1398670571

   @ijuma @showuon 
   
   can you please merge the patch?
   


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] fvaleri opened a new pull request, #13136: KAFKA-14582: Move JmxTool to tools

2023-01-20 Thread GitBox



fvaleri opened a new pull request, #13136:
URL: https://github.com/apache/kafka/pull/13136

   This PR is based on https://github.com/apache/kafka/pull/13131.
   
   This class is also used by the system tests, so I need to check if the 
replacement works fine there too.


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] fvaleri commented on a diff in pull request #13131: KAFKA-14628: Move CommandLineUtils and CommandDefaultOptions shared classes

2023-01-20 Thread GitBox



fvaleri commented on code in PR #13131:
URL: https://github.com/apache/kafka/pull/13131#discussion_r1082712140


##
server-common/src/main/java/org/apache/kafka/server/util/CommandLineUtils.java:
##
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.util;
+
+import joptsimple.OptionParser;
+import joptsimple.OptionSet;
+import joptsimple.OptionSpec;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.common.utils.Exit;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * Helper functions for dealing with command line utilities.
+ */
+public class CommandLineUtils {
+/**
+ * Check if there are no options or `--help` option from command line.
+ *
+ * @param commandOpts Acceptable options for a command
+ * @return true on matching the help check condition
+ */
+public static boolean isPrintHelpNeeded(CommandDefaultOptions commandOpts) 
{
+return commandOpts.args.length == 0 || 
commandOpts.options.has(commandOpts.helpOpt);
+}
+
+/**
+ * Check if there is `--version` option from command line.
+ *
+ * @param commandOpts Acceptable options for a command
+ * @return true on matching the help check condition
+ */
+public static boolean isPrintVersionNeeded(CommandDefaultOptions 
commandOpts) {
+return commandOpts.options.has(commandOpts.versionOpt);
+}
+
+/**
+ * Check and print help message if there is no options or `--help` option
+ * from command line, if `--version` is specified on the command line
+ * print version information and exit.
+ * NOTE: The function name is not strictly speaking correct anymore
+ * as it also checks whether the version needs to be printed, but
+ * refactoring this would have meant changing all command line tools
+ * and unnecessarily increased the blast radius of this change.
+ *
+ * @param commandOpts Acceptable options for a command
+ * @param message Message to display on successful check
+ */
+public static void printHelpAndExitIfNeeded(CommandDefaultOptions 
commandOpts, String message) {
+if (isPrintHelpNeeded(commandOpts)) {
+printUsageAndDie(commandOpts.parser, message);
+}
+if (isPrintVersionNeeded(commandOpts)) {
+printVersionAndDie();
+}
+}
+
+/**
+ * Check that all the listed options are present.
+ */
+public static void checkRequiredArgs(OptionParser parser, OptionSet 
options, OptionSpec... requiredList) {
+for (OptionSpec arg : requiredList) {
+if (!options.has(arg)) {
+printUsageAndDie(parser, String.format("Missing required 
argument \"%s\"", arg));
+}
+}
+}
+
+/**
+ * Check that none of the listed options are present.
+ */
+public static void checkInvalidArgs(OptionParser parser,
+OptionSet options,
+OptionSpec usedOption,
+OptionSpec... invalidOptions) {
+if (options.has(usedOption)) {
+for (OptionSpec arg : invalidOptions) {
+if (options.has(arg)) {
+printUsageAndDie(parser, String.format("Option \"%s\" 
can't be used with option \"%s\"", usedOption, arg));
+}
+}
+}
+}
+
+/**
+ * Check that none of the listed options are present.
+ */
+public static void checkInvalidArgs(OptionParser parser,
+OptionSet options,
+OptionSpec usedOption,
+Set> invalidOptions) {
+OptionSpec[] array = new OptionSpec[invalidOptions.size()];
+invalidOptions.toArray(array);
+checkInvalidArgs(parser, options, usedOption, array);
+}
+
+/**
+ * Check that none of the listed options are present with the combination 
of used options.
+ */
+public static void

[GitHub] [kafka] clolov commented on a diff in pull request #13131: KAFKA-14628: Move CommandLineUtils and CommandDefaultOptions shared classes

2023-01-20 Thread GitBox



clolov commented on code in PR #13131:
URL: https://github.com/apache/kafka/pull/13131#discussion_r1082663006


##
server-common/src/main/java/org/apache/kafka/server/util/CommandLineUtils.java:
##
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.util;
+
+import joptsimple.OptionParser;
+import joptsimple.OptionSet;
+import joptsimple.OptionSpec;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.common.utils.Exit;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * Helper functions for dealing with command line utilities.
+ */
+public class CommandLineUtils {
+/**
+ * Check if there are no options or `--help` option from command line.
+ *
+ * @param commandOpts Acceptable options for a command
+ * @return true on matching the help check condition
+ */
+public static boolean isPrintHelpNeeded(CommandDefaultOptions commandOpts) 
{
+return commandOpts.args.length == 0 || 
commandOpts.options.has(commandOpts.helpOpt);
+}
+
+/**
+ * Check if there is `--version` option from command line.
+ *
+ * @param commandOpts Acceptable options for a command
+ * @return true on matching the help check condition
+ */
+public static boolean isPrintVersionNeeded(CommandDefaultOptions 
commandOpts) {
+return commandOpts.options.has(commandOpts.versionOpt);
+}
+
+/**
+ * Check and print help message if there is no options or `--help` option
+ * from command line, if `--version` is specified on the command line
+ * print version information and exit.
+ * NOTE: The function name is not strictly speaking correct anymore
+ * as it also checks whether the version needs to be printed, but
+ * refactoring this would have meant changing all command line tools
+ * and unnecessarily increased the blast radius of this change.
+ *
+ * @param commandOpts Acceptable options for a command
+ * @param message Message to display on successful check
+ */
+public static void printHelpAndExitIfNeeded(CommandDefaultOptions 
commandOpts, String message) {
+if (isPrintHelpNeeded(commandOpts)) {
+printUsageAndDie(commandOpts.parser, message);
+}
+if (isPrintVersionNeeded(commandOpts)) {
+printVersionAndDie();
+}
+}
+
+/**
+ * Check that all the listed options are present.
+ */
+public static void checkRequiredArgs(OptionParser parser, OptionSet 
options, OptionSpec... requiredList) {
+for (OptionSpec arg : requiredList) {
+if (!options.has(arg)) {
+printUsageAndDie(parser, String.format("Missing required 
argument \"%s\"", arg));
+}
+}
+}
+
+/**
+ * Check that none of the listed options are present.
+ */
+public static void checkInvalidArgs(OptionParser parser,
+OptionSet options,
+OptionSpec usedOption,
+OptionSpec... invalidOptions) {
+if (options.has(usedOption)) {
+for (OptionSpec arg : invalidOptions) {
+if (options.has(arg)) {
+printUsageAndDie(parser, String.format("Option \"%s\" 
can't be used with option \"%s\"", usedOption, arg));
+}
+}
+}
+}
+
+/**
+ * Check that none of the listed options are present.
+ */
+public static void checkInvalidArgs(OptionParser parser,
+OptionSet options,
+OptionSpec usedOption,
+Set> invalidOptions) {
+OptionSpec[] array = new OptionSpec[invalidOptions.size()];
+invalidOptions.toArray(array);
+checkInvalidArgs(parser, options, usedOption, array);
+}
+
+/**
+ * Check that none of the listed options are present with the combination 
of used options.
+ */
+public static void

[GitHub] [kafka] ijuma commented on a diff in pull request #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox



ijuma commented on code in PR #13135:
URL: https://github.com/apache/kafka/pull/13135#discussion_r1082623877


##
clients/src/main/java/org/apache/kafka/common/record/CompressionType.java:
##
@@ -126,6 +144,11 @@ public OutputStream wrapForOutput(ByteBufferOutputStream 
buffer, byte messageVer
 public InputStream wrapForInput(ByteBuffer buffer, byte 
messageVersion, BufferSupplier decompressionBufferSupplier) {
 return ZstdFactory.wrapForInput(buffer, messageVersion, 
decompressionBufferSupplier);
 }
+
+@Override
+public int getRecommendedDOutSize() {
+return 16 * 1024; // 16KB

Review Comment:
   OK, that's fine.



-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] ijuma commented on a diff in pull request #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox



ijuma commented on code in PR #13135:
URL: https://github.com/apache/kafka/pull/13135#discussion_r1082622801


##
clients/src/main/java/org/apache/kafka/common/record/CompressionType.java:
##
@@ -126,6 +144,11 @@ public OutputStream wrapForOutput(ByteBufferOutputStream 
buffer, byte messageVer
 public InputStream wrapForInput(ByteBuffer buffer, byte 
messageVersion, BufferSupplier decompressionBufferSupplier) {
 return ZstdFactory.wrapForInput(buffer, messageVersion, 
decompressionBufferSupplier);
 }
+
+@Override
+public int getRecommendedDOutSize() {
+return 16 * 1024; // 16KB

Review Comment:
   The PR says:
   
   > we pushed the skipping of key/value logic to zstd-jni implementation 
instead of using the one provided by BufferedInputStream



-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] divijvaidya commented on a diff in pull request #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox



divijvaidya commented on code in PR #13135:
URL: https://github.com/apache/kafka/pull/13135#discussion_r1082615267


##
clients/src/main/java/org/apache/kafka/common/record/CompressionType.java:
##
@@ -126,6 +144,11 @@ public OutputStream wrapForOutput(ByteBufferOutputStream 
buffer, byte messageVer
 public InputStream wrapForInput(ByteBuffer buffer, byte 
messageVersion, BufferSupplier decompressionBufferSupplier) {
 return ZstdFactory.wrapForInput(buffer, messageVersion, 
decompressionBufferSupplier);
 }
+
+@Override
+public int getRecommendedDOutSize() {
+return 16 * 1024; // 16KB

Review Comment:
   This will be picked up as a separate follow-up PR: 
https://issues.apache.org/jira/browse/KAFKA-14634 
   
   I decided to keep it separate is to introduce fewer changes at a time and 
measure their impact.



-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox



ijuma commented on PR #13135:
URL: https://github.com/apache/kafka/pull/13135#issuecomment-1398459548

   One more thing: when it comes to the testing, can we include the case where 
the batches have a single 10 byte message?


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] ijuma commented on a diff in pull request #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox



ijuma commented on code in PR #13135:
URL: https://github.com/apache/kafka/pull/13135#discussion_r1082588469


##
clients/src/main/java/org/apache/kafka/common/compress/ZstdFactory.java:
##
@@ -62,10 +68,11 @@ public void release(ByteBuffer buffer) {
 }
 };
 
-// Set output buffer (uncompressed) to 16 KB (none by default) to 
ensure reasonable performance
-// in cases where the caller reads a small number of bytes 
(potentially a single byte).
-return new BufferedInputStream(new ZstdInputStreamNoFinalizer(new 
ByteBufferInputStream(buffer),
-bufferPool), 16 * 1024);
+// We do not use an intermediate buffer to store the decompressed 
data as a result of JNI read() calls using
+// `ZstdInputStreamNoFinalizer` here. Every read() call to 
`DataInputStream` will be a JNI call and the
+// caller is expected to balance the tradeoff between reading 
large amount of data vs. making multiple JNI
+// calls.
+return new DataInputStream(new ZstdInputStreamNoFinalizer(new 
ByteBufferInputStream(buffer), bufferPool));

Review Comment:
   2 questions:
   1. Why do we wrap into DataInputStream?
   2. Have as checked that there are no workloads where we end up doing too 
many JNI calls?



##
clients/src/main/java/org/apache/kafka/common/record/CompressionType.java:
##
@@ -47,6 +47,11 @@ public OutputStream wrapForOutput(ByteBufferOutputStream 
buffer, byte messageVer
 public InputStream wrapForInput(ByteBuffer buffer, byte 
messageVersion, BufferSupplier decompressionBufferSupplier) {
 return new ByteBufferInputStream(buffer);
 }
+
+@Override
+public int getRecommendedDOutSize() {
+return 2 * 1024; // 2KB

Review Comment:
   What's the meaning of this for an uncompressed stream?



##
clients/src/main/java/org/apache/kafka/common/compress/ZstdFactory.java:
##
@@ -26,21 +26,25 @@
 import org.apache.kafka.common.utils.ByteBufferInputStream;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
 
-import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
+import java.io.DataInputStream;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 
 public class ZstdFactory {
+/**
+ * Default compression level
+ */
+private static final int DEFAULT_COMPRESSION_LEVEL = 3;

Review Comment:
   Since this is unrelated, do we have to include it as part of this PR?



##
clients/src/main/java/org/apache/kafka/common/record/CompressionType.java:
##
@@ -126,6 +144,11 @@ public OutputStream wrapForOutput(ByteBufferOutputStream 
buffer, byte messageVer
 public InputStream wrapForInput(ByteBuffer buffer, byte 
messageVersion, BufferSupplier decompressionBufferSupplier) {
 return ZstdFactory.wrapForInput(buffer, messageVersion, 
decompressionBufferSupplier);
 }
+
+@Override
+public int getRecommendedDOutSize() {
+return 16 * 1024; // 16KB

Review Comment:
   We decided not to get this info from the zstd library?



##
clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java:
##
@@ -273,20 +272,32 @@ public int partitionLeaderEpoch() {
 public DataInputStream recordInputStream(BufferSupplier bufferSupplier) {
 final ByteBuffer buffer = this.buffer.duplicate();
 buffer.position(RECORDS_OFFSET);
-return new DataInputStream(compressionType().wrapForInput(buffer, 
magic(), bufferSupplier));
+final InputStream decompressedStream = 
compressionType().wrapForInput(buffer, magic(), bufferSupplier);
+return decompressedStream instanceof DataInputStream ? 
(DataInputStream) decompressedStream : new DataInputStream(decompressedStream);
 }
 
 private CloseableIterator compressedIterator(BufferSupplier 
bufferSupplier, boolean skipKeyValue) {
 final DataInputStream inputStream = recordInputStream(bufferSupplier);
 
 if (skipKeyValue) {
 // this buffer is used to skip length delimited fields like key, 
value, headers
-byte[] skipArray = new byte[MAX_SKIP_BUFFER_SIZE];
+final ByteBuffer skipBuffer = 
bufferSupplier.get(compressionType().getRecommendedDOutSize());

Review Comment:
   I thought we wanted to call the underlying skipBytes API versus doing the 
skipping by reading into a skip buffer. I don't see that change. What am I 
missing?



-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] divijvaidya commented on pull request #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox



divijvaidya commented on PR #13135:
URL: https://github.com/apache/kafka/pull/13135#issuecomment-1398428117

   @ijuma please review when you get a chance since you already have context 
about this code change.


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] divijvaidya commented on a diff in pull request #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox



divijvaidya commented on code in PR #13135:
URL: https://github.com/apache/kafka/pull/13135#discussion_r1082366873


##
clients/src/main/java/org/apache/kafka/common/compress/ZstdFactory.java:
##
@@ -26,21 +26,25 @@
 import org.apache.kafka.common.utils.ByteBufferInputStream;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
 
-import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
+import java.io.DataInputStream;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 
 public class ZstdFactory {
+/**
+ * Default compression level
+ */
+private static final int DEFAULT_COMPRESSION_LEVEL = 3;

Review Comment:
   FYI reviewer
   
   This change is a no-op since we are already using the default value of 3 
when no value is provided. This change has been made to make it explicit in 
Kafka code that we are using compression level of 3 with zstd.



-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] clolov commented on pull request #12821: KAFKA-14132: Replace PowerMock and EasyMock with Mockito in streams tests

2023-01-20 Thread GitBox



clolov commented on PR #12821:
URL: https://github.com/apache/kafka/pull/12821#issuecomment-1398172190

   Thank you very much for the review and merge!


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] divijvaidya opened a new pull request, #13135: KAFKA-14633: Reduce data copy & buffer allocation during decompression

2023-01-20 Thread GitBox

divijvaidya opened a new pull request, #13135:
URL: https://github.com/apache/kafka/pull/13135

This covers two JIRAs https://issues.apache.org/jira/browse/KAFKA-14632 and
https://issues.apache.org/jira/browse/KAFKA-14633

## Background
![Screenshot 2023-01-19 at 18 27
45](https://user-images.githubusercontent.com/71267/213521204-bb3228ed-7d21-4e07-a520-697ea6fcc0ed.png)
Currently, we use 2 intermediate buffers while handling decompressed data
(one of size 2KB and another of size 16KB). These buffers are (de)allocated
once per batch.

The impact of this was observed in a flamegraph analysis for a compressed
workload where we observed that 75% of CPU during `appendAsLeader()` is taken
up by `ValidateMessagesAndAssignOffsets`.

![Screenshot 2023-01-20 at 10 41
08](https://user-images.githubusercontent.com/71267/213664252-389eaf3d-b8aa-465b-b010-db1024663d6f.png)

## Change
With this PR:
1. we are removing the number of intermediate buffers from 2 to 1. This
reduces 1 point of data copy. Note that this removed data copy occurred in
chunks of 2kb at a time, multiple times. This is achieved by getting rid of
`BufferedInputStream` and moving to `DataInputStream`. This change has only
been made for `zstd` and `gzip`.
2. we are using thread local buffer pool for both the buffers involved in
the process of decompression. This change impacts all compression types.
3. we pushed the skipping of key/value logic to

After the change, the above buffer allocation looks as follows:
![Screenshot 2023-01-19 at 18 28
14](https://user-images.githubusercontent.com/71267/213525653-917ac5ee-810a-435e-bf84-c97d6b76005e.png)

## Results
After this change, a JMH benchmark for `ValidateMessagesAndAssignOffsets`
demonstrated 10-50% increased throughput across all compression types without
any regression. The improvement is prominent when thread cached buffer pools
are used with 1-2% regression in some limited scenarios.

When buffer pools are not used (NO_CACHING in the results), we observed GZIP
having 10% better performance in some cases with 1-4% regression for some other
scenarios. Overall, without using the buffer pools, the upside of this code
change is limited to single digit improvements in certain scenarios.

Details results from JMH benchmark are available here:
[benchmark-jira.xlsx](https://github.com/apache/kafka/files/10465049/benchmark-jira.xlsx)

## Testing
- Sanity testing using the existing unit test to ensure that we don't impact
correctness.
- JMH benchmarks for all compression types to ensure that we did not regress
other compression types.

--
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka] ableegoldman opened a new pull request, #13134: MINOR: fix flaky StickyAssignorTest.testLargeAssignmentAndGroupWithNonEqualSubscription

2023-01-20 Thread GitBox



ableegoldman opened a new pull request, #13134:
URL: https://github.com/apache/kafka/pull/13134

   This test is supposed to be a sanity check that rebalancing with a large 
number of partitions/consumers won't start to take obscenely long or approach 
the `max.poll.interval.ms` -- bumping up the timeout by another 30s still feels 
very reasonable considering the test is for 1 million partitions


-- 
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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

1 2 3 4 5 6 7 8 9 10 >

1 - 100 of 51425 matches

Mail list logo