You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ab...@apache.org on 2022/06/14 05:27:19 UTC
[kafka] branch trunk updated: HOTFIX: null check keys of ProducerRecord when computing sizeInBytes (#12288)
This is an automated email from the ASF dual-hosted git repository.
ableegoldman 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 3189a8648f HOTFIX: null check keys of ProducerRecord when computing sizeInBytes (#12288)
3189a8648f is described below
commit 3189a8648f3abf3e16c5932dfa37e0d28d97a016
Author: A. Sophie Blee-Goldman <so...@confluent.io>
AuthorDate: Mon Jun 13 22:27:06 2022 -0700
HOTFIX: null check keys of ProducerRecord when computing sizeInBytes (#12288)
Minor followup to #12235 that adds a null check on the record key in the new ClientUtils#producerRecordSizeInBytes utility method, as there are valid cases in which we might be sending records with null keys to the Producer, such as a simple builder.stream("non-keyed-input-topic").filter(...).to("output-topic")
Reviewers: Matthias J. Sax <ma...@confluent.io>, Bill Bejeck <bi...@confluent.io>
---
.../streams/processor/internals/ClientUtils.java | 10 ++---
.../processor/internals/ClientUtilsTest.java | 44 +++++++++++++++++++++-
.../org/apache/kafka/streams/TestTopicsTest.java | 12 ++----
3 files changed, 50 insertions(+), 16 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ClientUtils.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ClientUtils.java
index ea48c64617..1177e29d82 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ClientUtils.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ClientUtils.java
@@ -174,7 +174,7 @@ public class ClientUtils {
public static long producerRecordSizeInBytes(final ProducerRecord<byte[], byte[]> record) {
return recordSizeInBytes(
- record.key().length,
+ record.key() == null ? 0 : record.key().length,
record.value() == null ? 0 : record.value().length,
record.topic(),
record.headers()
@@ -190,10 +190,10 @@ public class ClientUtils {
);
}
- public static long recordSizeInBytes(final long keyBytes,
- final long valueBytes,
- final String topic,
- final Headers headers) {
+ private static long recordSizeInBytes(final long keyBytes,
+ final long valueBytes,
+ final String topic,
+ final Headers headers) {
long headerSizeInBytes = 0L;
if (headers != null) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ClientUtilsTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ClientUtilsTest.java
index 775268496e..d715a3f975 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ClientUtilsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ClientUtilsTest.java
@@ -69,10 +69,11 @@ public class ClientUtilsTest {
private static final Headers HEADERS = new RecordHeaders(asList(
new RecordHeader("h1", "headerVal1".getBytes()), // 2 + 10 --> 12 bytes
- new RecordHeader("h2", "headerVal2".getBytes())
- )); // 2 + 10 --> 12 bytes
+ new RecordHeader("h2", "headerVal2".getBytes()) // 2 + 10 --> 12 bytes
+ ));
private static final int HEADERS_BYTES = 24;
+ // 20 bytes
private static final int RECORD_METADATA_BYTES =
8 + // timestamp
8 + // offset
@@ -86,6 +87,14 @@ public class ClientUtilsTest {
HEADERS_BYTES +
RECORD_METADATA_BYTES;
+ // 54 bytes
+ private static final long NULL_KEY_SIZE_IN_BYTES =
+ VALUE_BYTES +
+ TOPIC_BYTES +
+ HEADERS_BYTES +
+ RECORD_METADATA_BYTES;
+
+ // 52 bytes
private static final long TOMBSTONE_SIZE_IN_BYTES =
KEY_BYTES +
TOPIC_BYTES +
@@ -202,6 +211,37 @@ public class ClientUtilsTest {
assertThat(producerRecordSizeInBytes(record), equalTo(SIZE_IN_BYTES));
}
+ @Test
+ public void shouldComputeSizeInBytesForConsumerRecordWithNullKey() {
+ final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>(
+ TOPIC,
+ 1,
+ 0,
+ 0L,
+ TimestampType.CREATE_TIME,
+ 0,
+ 5,
+ null,
+ VALUE,
+ HEADERS,
+ Optional.empty()
+ );
+ assertThat(consumerRecordSizeInBytes(record), equalTo(NULL_KEY_SIZE_IN_BYTES));
+ }
+
+ @Test
+ public void shouldComputeSizeInBytesForProducerRecordWithNullKey() {
+ final ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
+ TOPIC,
+ 1,
+ 0L,
+ null,
+ VALUE,
+ HEADERS
+ );
+ assertThat(producerRecordSizeInBytes(record), equalTo(NULL_KEY_SIZE_IN_BYTES));
+ }
+
@Test
public void shouldComputeSizeInBytesForConsumerRecordWithNullValue() {
final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>(
diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java
index f39b2b554a..84b8c5ebd6 100644
--- a/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java
+++ b/streams/test-utils/src/test/java/org/apache/kafka/streams/TestTopicsTest.java
@@ -31,8 +31,6 @@ import org.apache.kafka.streams.test.TestRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.time.Instant;
@@ -55,8 +53,6 @@ import static org.hamcrest.Matchers.hasProperty;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class TestTopicsTest {
- private static final Logger log = LoggerFactory.getLogger(TestTopicsTest.class);
-
private final static String INPUT_TOPIC = "input";
private final static String OUTPUT_TOPIC = "output1";
private final static String INPUT_TOPIC_MAP = OUTPUT_TOPIC;
@@ -171,15 +167,13 @@ public class TestTopicsTest {
}
@Test
- public void testPipeInputWithNullKey() {
+ public void testKeyValuesToMapWithNull() {
final TestInputTopic<Long, String> inputTopic =
testDriver.createInputTopic(INPUT_TOPIC, longSerde.serializer(), stringSerde.serializer());
final TestOutputTopic<Long, String> outputTopic =
testDriver.createOutputTopic(OUTPUT_TOPIC, longSerde.deserializer(), stringSerde.deserializer());
- final StreamsException exception = assertThrows(StreamsException.class, () -> inputTopic.pipeInput("value"));
- assertThat(exception.getCause() instanceof NullPointerException, is(true));
- assertThat(outputTopic.readKeyValuesToMap().isEmpty(), is(true));
-
+ inputTopic.pipeInput("value");
+ assertThrows(IllegalStateException.class, outputTopic::readKeyValuesToMap);
}
@Test