Skip to content

KAFKA-17872: Update consumed offsets on records with invalid timestamp #17710

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 2 commits into from
Nov 10, 2024
Merged

KAFKA-17872: Update consumed offsets on records with invalid timestamp #17710

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
32a2a86
KAFKA-17872: Update consumed offsets on records with invalid timestamp
mjsax Nov 7, 2024
6543e2e
review comments
mjsax Nov 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,7 @@ private void updateHead() {
deserialized.topic(), deserialized.partition(), deserialized.offset(), timestamp, timestampExtractor.getClass().getCanonicalName()
);
droppedRecordsSensor.record();
lastCorruptedRecord = raw;

@mjsax mjsax Nov 7, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the actual fix -- we track the record with invalid ts as "corrupted record" instead of dropping it completely. The CorrecuptedRecrods allows us to update the offset later, but would be dropped by itself and not be processed.

continue;
}
headRecord = new StampedRecord(deserialized, timestamp);
Expand Down
59 changes: 36 additions & 23 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordQueueTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.LogContext;
Expand Down Expand Up @@ -74,8 +74,7 @@ public class RecordQueueTest {
private final StreamsMetricsImpl streamsMetrics =
new StreamsMetricsImpl(metrics, "mock", new MockTime());

@SuppressWarnings("rawtypes")
final InternalMockProcessorContext context = new InternalMockProcessorContext<>(
final InternalMockProcessorContext<Integer, Integer> context = new InternalMockProcessorContext<>(

@mjsax mjsax Nov 7, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Side cleanup to fix warnings

StateSerdes.withBuiltinTypes("anyName", Bytes.class, Bytes.class),
new MockRecordCollector(),
metrics
Expand All @@ -88,19 +87,28 @@ public class RecordQueueTest {
timestampExtractor,
new LogAndFailExceptionHandler(),
context,
new LogContext());
new LogContext()
);
private final RecordQueue queueThatSkipsDeserializeErrors = new RecordQueue(
new TopicPartition("topic", 1),
mockSourceNodeWithMetrics,
timestampExtractor,
new LogAndContinueExceptionHandler(),
context,
new LogContext());
new LogContext()
);
private final RecordQueue queueThatSkipsInvalidTimestamps = new RecordQueue(
new TopicPartition("topic", 1),
mockSourceNodeWithMetrics,
new LogAndSkipOnInvalidTimestamp(),
new LogAndFailExceptionHandler(),
context,
new LogContext()
);

private final byte[] recordValue = intSerializer.serialize(null, 10);
private final byte[] recordKey = intSerializer.serialize(null, 1);

@SuppressWarnings("unchecked")
@BeforeEach
public void before() {
mockSourceNodeWithMetrics.init(context);
Expand Down Expand Up @@ -340,7 +348,7 @@ public void shouldSetTimestampAndRespectMaxTimestampPolicy() {

@Test
public void shouldThrowStreamsExceptionWhenKeyDeserializationFails() {
final byte[] key = Serdes.Long().serializer().serialize("foo", 1L);
final byte[] key = new LongSerializer().serialize("foo", 1L);

@mjsax mjsax Nov 7, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Side cleanup

final List<ConsumerRecord<byte[], byte[]>> records = Collections.singletonList(
new ConsumerRecord<>("topic", 1, 1, 0L, TimestampType.CREATE_TIME, 0, 0, key, recordValue,
new RecordHeaders(), Optional.empty()));
Expand All @@ -354,7 +362,7 @@ public void shouldThrowStreamsExceptionWhenKeyDeserializationFails() {

@Test
public void shouldThrowStreamsExceptionWhenValueDeserializationFails() {
final byte[] value = Serdes.Long().serializer().serialize("foo", 1L);
final byte[] value = new LongSerializer().serialize("foo", 1L);
final List<ConsumerRecord<byte[], byte[]>> records = Collections.singletonList(
new ConsumerRecord<>("topic", 1, 1, 0L, TimestampType.CREATE_TIME, 0, 0, recordKey, value,
new RecordHeaders(), Optional.empty()));
Expand All @@ -368,7 +376,7 @@ public void shouldThrowStreamsExceptionWhenValueDeserializationFails() {

@Test
public void shouldNotThrowStreamsExceptionWhenKeyDeserializationFailsWithSkipHandler() {
final byte[] key = Serdes.Long().serializer().serialize("foo", 1L);
final byte[] key = new LongSerializer().serialize("foo", 1L);
final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>("topic", 1, 1, 0L,
TimestampType.CREATE_TIME, 0, 0, key, recordValue,
new RecordHeaders(), Optional.empty());
Expand All @@ -381,7 +389,7 @@ public void shouldNotThrowStreamsExceptionWhenKeyDeserializationFailsWithSkipHan

@Test
public void shouldNotThrowStreamsExceptionWhenValueDeserializationFailsWithSkipHandler() {
final byte[] value = Serdes.Long().serializer().serialize("foo", 1L);
final byte[] value = new LongSerializer().serialize("foo", 1L);
final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>("topic", 1, 1, 0L,
TimestampType.CREATE_TIME, 0, 0, recordKey, value,
new RecordHeaders(), Optional.empty());
Expand All @@ -404,7 +412,7 @@ public void shouldThrowOnNegativeTimestamp() {
mockSourceNodeWithMetrics,
new FailOnInvalidTimestamp(),
new LogAndContinueExceptionHandler(),
new InternalMockProcessorContext(),
new InternalMockProcessorContext<>(),
new LogContext());

final StreamsException exception = assertThrows(
Expand All @@ -421,20 +429,25 @@ public void shouldThrowOnNegativeTimestamp() {

@Test
public void shouldDropOnNegativeTimestamp() {
final List<ConsumerRecord<byte[], byte[]>> records = Collections.singletonList(
new ConsumerRecord<>("topic", 1, 1, -1L, TimestampType.CREATE_TIME, 0, 0, recordKey, recordValue,
new RecordHeaders(), Optional.empty()));
final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>(
"topic",
1,
1,
-1L, // negative timestamp
TimestampType.CREATE_TIME,
0,
0,
recordKey,
recordValue,
new RecordHeaders(),
Optional.empty()
);
final List<ConsumerRecord<byte[], byte[]>> records = Collections.singletonList(record);

final RecordQueue queue = new RecordQueue(
new TopicPartition("topic", 1),
mockSourceNodeWithMetrics,
new LogAndSkipOnInvalidTimestamp(),
new LogAndContinueExceptionHandler(),
new InternalMockProcessorContext(),
new LogContext());
queue.addRawRecords(records);
queueThatSkipsInvalidTimestamps.addRawRecords(records);

assertEquals(0, queue.size());
assertEquals(1, queueThatSkipsInvalidTimestamps.size());
assertEquals(new CorruptedRecord(record), queueThatSkipsInvalidTimestamps.poll(0));
}

@Test
Expand Down
Loading