You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2024/11/10 01:45:09 UTC
(kafka) branch 3.9 updated: KAFKA-17872: Update consumed offsets on records with invalid timestamp (#17710)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new 587c3cfec5d KAFKA-17872: Update consumed offsets on records with invalid timestamp (#17710)
587c3cfec5d is described below
commit 587c3cfec5dd3b2b408ef4e6ed75878f56daf15d
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Sat Nov 9 17:21:16 2024 -0800
KAFKA-17872: Update consumed offsets on records with invalid timestamp (#17710)
TimestampExtractor allows to drop records by returning a timestamp of -1. For this case, we still need to update consumed offsets to allows us to commit progress.
Reviewers: Bill Bejeck <bi...@confluent.io>
---
.../streams/processor/internals/RecordQueue.java | 1 +
.../processor/internals/RecordQueueTest.java | 59 ++--
.../processor/internals/StreamTaskTest.java | 341 +++++++++++++++------
3 files changed, 282 insertions(+), 119 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
index a6b30a07ef9..784ed3042ea 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
@@ -228,6 +228,7 @@ public class RecordQueue {
deserialized.topic(), deserialized.partition(), deserialized.offset(), timestamp, timestampExtractor.getClass().getCanonicalName()
);
droppedRecordsSensor.record();
+ lastCorruptedRecord = raw;
continue;
}
headRecord = new StampedRecord(deserialized, timestamp);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordQueueTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordQueueTest.java
index ad6a33d690d..6654f146e2a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordQueueTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordQueueTest.java
@@ -27,7 +27,7 @@ import org.apache.kafka.common.record.TimestampType;
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;
@@ -75,8 +75,7 @@ public class RecordQueueTest {
private final StreamsMetricsImpl streamsMetrics =
new StreamsMetricsImpl(metrics, "mock", StreamsConfig.METRICS_LATEST, new MockTime());
- @SuppressWarnings("rawtypes")
- final InternalMockProcessorContext context = new InternalMockProcessorContext<>(
+ final InternalMockProcessorContext<Integer, Integer> context = new InternalMockProcessorContext<>(
StateSerdes.withBuiltinTypes("anyName", Bytes.class, Bytes.class),
new MockRecordCollector(),
metrics
@@ -89,19 +88,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);
@@ -328,7 +336,7 @@ public class RecordQueueTest {
@Test
public void shouldThrowStreamsExceptionWhenKeyDeserializationFails() {
- final byte[] key = Serdes.Long().serializer().serialize("foo", 1L);
+ final byte[] key = new LongSerializer().serialize("foo", 1L);
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()));
@@ -342,7 +350,7 @@ public class RecordQueueTest {
@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()));
@@ -356,7 +364,7 @@ public class RecordQueueTest {
@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());
@@ -369,7 +377,7 @@ public class RecordQueueTest {
@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());
@@ -392,7 +400,7 @@ public class RecordQueueTest {
mockSourceNodeWithMetrics,
new FailOnInvalidTimestamp(),
new LogAndContinueExceptionHandler(),
- new InternalMockProcessorContext(),
+ new InternalMockProcessorContext<>(),
new LogContext());
final StreamsException exception = assertThrows(
@@ -409,20 +417,25 @@ public class RecordQueueTest {
@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
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index cecfdd35ca1..2650f708652 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -38,14 +38,15 @@ import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.CumulativeSum;
import org.apache.kafka.common.record.TimestampType;
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.Serializer;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TopologyConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.errors.ErrorHandlerContext;
import org.apache.kafka.streams.errors.LockException;
import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
@@ -60,10 +61,13 @@ import org.apache.kafka.streams.errors.TaskCorruptedException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.errors.internals.FailedProcessingException;
+import org.apache.kafka.streams.processor.FailOnInvalidTimestamp;
+import org.apache.kafka.streams.processor.LogAndSkipOnInvalidTimestamp;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.TimestampExtractor;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.Task.TaskType;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
@@ -88,7 +92,6 @@ import org.mockito.quality.Strictness;
import java.io.File;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
@@ -159,12 +162,12 @@ public class StreamTaskTest {
private final TopicPartition partition1 = new TopicPartition(topic1, 1);
private final TopicPartition partition2 = new TopicPartition(topic2, 1);
private final Set<TopicPartition> partitions = mkSet(partition1, partition2);
- private final Serializer<Integer> intSerializer = Serdes.Integer().serializer();
- private final Deserializer<Integer> intDeserializer = Serdes.Integer().deserializer();
+ private final Serializer<Integer> intSerializer = new IntegerSerializer();
+ private final Deserializer<Integer> intDeserializer = new IntegerDeserializer();
private final MockSourceNode<Integer, Integer> source1 = new MockSourceNode<>(intDeserializer, intDeserializer);
private final MockSourceNode<Integer, Integer> source2 = new MockSourceNode<>(intDeserializer, intDeserializer);
- private final MockSourceNode<Integer, Integer> source3 = new MockSourceNode<Integer, Integer>(intDeserializer, intDeserializer) {
+ private final MockSourceNode<Integer, Integer> source3 = new MockSourceNode<>(intDeserializer, intDeserializer) {
@Override
public void process(final Record<Integer, Integer> record) {
throw new RuntimeException("KABOOM!");
@@ -175,7 +178,7 @@ public class StreamTaskTest {
throw new RuntimeException("KABOOM!");
}
};
- private final MockSourceNode<Integer, Integer> timeoutSource = new MockSourceNode<Integer, Integer>(intDeserializer, intDeserializer) {
+ private final MockSourceNode<Integer, Integer> timeoutSource = new MockSourceNode<>(intDeserializer, intDeserializer) {
@Override
public void process(final Record<Integer, Integer> record) {
throw new TimeoutException("Kaboom!");
@@ -250,18 +253,54 @@ public class StreamTaskTest {
}
private static StreamsConfig createConfig(final String eosConfig, final String enforcedProcessingValue) {
- return createConfig(eosConfig, enforcedProcessingValue, LogAndFailExceptionHandler.class.getName(), LogAndFailProcessingExceptionHandler.class.getName());
+ return createConfig(
+ eosConfig,
+ enforcedProcessingValue,
+ LogAndFailExceptionHandler.class,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class
+ );
+ }
+
+ private static StreamsConfig createConfig(final Class<? extends DeserializationExceptionHandler> deserializationExceptionHandler) {
+ return createConfig(
+ AT_LEAST_ONCE,
+ "0", // max.task.idle.ms
+ deserializationExceptionHandler,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class
+ );
}
- private static StreamsConfig createConfig(final String eosConfig, final String enforcedProcessingValue, final String deserializationExceptionHandler) {
- return createConfig(eosConfig, enforcedProcessingValue, deserializationExceptionHandler, LogAndFailProcessingExceptionHandler.class.getName());
+ private static StreamsConfig createConfigWithTsExtractor(final Class<? extends TimestampExtractor> timestampExtractor) {
+ return createConfig(
+ AT_LEAST_ONCE,
+ "0", // max.task.idle.ms
+ LogAndFailExceptionHandler.class,
+ LogAndFailProcessingExceptionHandler.class,
+ timestampExtractor
+ );
+ }
+
+ private static StreamsConfig createConfig(
+ final String enforcedProcessingValue,
+ final Class<? extends ProcessingExceptionHandler> processingExceptionHandler
+ ) {
+ return createConfig(
+ AT_LEAST_ONCE,
+ enforcedProcessingValue,
+ LogAndFailExceptionHandler.class,
+ processingExceptionHandler,
+ FailOnInvalidTimestamp.class
+ );
}
private static StreamsConfig createConfig(
final String eosConfig,
final String enforcedProcessingValue,
- final String deserializationExceptionHandler,
- final String processingExceptionHandler) {
+ final Class<? extends DeserializationExceptionHandler> deserializationExceptionHandler,
+ final Class<? extends ProcessingExceptionHandler> processingExceptionHandler,
+ final Class<? extends TimestampExtractor> timestampExtractor) {
final String canonicalPath;
try {
canonicalPath = BASE_DIR.getCanonicalPath();
@@ -277,8 +316,9 @@ public class StreamTaskTest {
mkEntry(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MockTimestampExtractor.class.getName()),
mkEntry(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig),
mkEntry(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, enforcedProcessingValue),
- mkEntry(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, deserializationExceptionHandler),
- mkEntry(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, processingExceptionHandler)
+ mkEntry(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, deserializationExceptionHandler.getName()),
+ mkEntry(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, processingExceptionHandler.getName()),
+ mkEntry(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, timestampExtractor.getName())
)));
}
@@ -402,36 +442,38 @@ public class StreamTaskTest {
task.addPartitionsForOffsetReset(Collections.singleton(partition1));
final AtomicReference<AssertionError> shouldNotSeek = new AtomicReference<>();
- final MockConsumer<byte[], byte[]> consumer = new MockConsumer<byte[], byte[]>(OffsetResetStrategy.EARLIEST) {
- @Override
- public void seek(final TopicPartition partition, final long offset) {
- final AssertionError error = shouldNotSeek.get();
- if (error != null) {
- throw error;
+ try (final MockConsumer<byte[], byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST) {
+ @Override
+ public void seek(final TopicPartition partition, final long offset) {
+ final AssertionError error = shouldNotSeek.get();
+ if (error != null) {
+ throw error;
+ }
+ super.seek(partition, offset);
}
- super.seek(partition, offset);
- }
- };
- consumer.assign(asList(partition1, partition2));
- consumer.updateBeginningOffsets(mkMap(mkEntry(partition1, 0L), mkEntry(partition2, 0L)));
+ }) {
- consumer.seek(partition1, 5L);
- consumer.seek(partition2, 15L);
+ consumer.assign(asList(partition1, partition2));
+ consumer.updateBeginningOffsets(mkMap(mkEntry(partition1, 0L), mkEntry(partition2, 0L)));
- shouldNotSeek.set(new AssertionError("Should not seek"));
+ consumer.seek(partition1, 5L);
+ consumer.seek(partition2, 15L);
- // We need to keep a separate reference to the arguments of Consumer#accept
- // because the underlying data-structure is emptied and on verification time
- // it is reported as empty.
- final Set<TopicPartition> partitionsAtCall = new HashSet<>();
+ shouldNotSeek.set(new AssertionError("Should not seek"));
- task.initializeIfNeeded();
- task.completeRestoration(partitionsAtCall::addAll);
+ // We need to keep a separate reference to the arguments of Consumer#accept
+ // because the underlying data-structure is emptied and on verification time
+ // it is reported as empty.
+ final Set<TopicPartition> partitionsAtCall = new HashSet<>();
- // because we mocked the `resetter` positions don't change
- assertThat(consumer.position(partition1), equalTo(5L));
- assertThat(consumer.position(partition2), equalTo(15L));
- assertThat(partitionsAtCall, equalTo(Collections.singleton(partition1)));
+ task.initializeIfNeeded();
+ task.completeRestoration(partitionsAtCall::addAll);
+
+ // because we mocked the `resetter` positions don't change
+ assertThat(consumer.position(partition1), equalTo(5L));
+ assertThat(consumer.position(partition2), equalTo(15L));
+ assertThat(partitionsAtCall, equalTo(Collections.singleton(partition1)));
+ }
}
@Test
@@ -942,10 +984,9 @@ public class StreamTaskTest {
reporter.contextChange(metricsContext);
metrics.addReporter(reporter);
- final String threadIdTag = THREAD_ID_TAG;
assertTrue(reporter.containsMbean(String.format(
"kafka.streams:type=stream-task-metrics,%s=%s,task-id=%s",
- threadIdTag,
+ THREAD_ID_TAG,
threadId,
task.id()
)));
@@ -1469,18 +1510,15 @@ public class StreamTaskTest {
assertFalse(task.process(0L));
- final byte[] bytes = ByteBuffer.allocate(4).putInt(1).array();
-
- task.addRecords(partition1, singleton(new ConsumerRecord<>(topic1, 1, 0, bytes, bytes)));
+ task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
assertFalse(task.process(0L));
assertThat("task is idling", task.timeCurrentIdlingStarted().isPresent());
- task.addRecords(partition2, singleton(new ConsumerRecord<>(topic2, 1, 0, bytes, bytes)));
+ task.addRecords(partition2, singleton(getConsumerRecordWithOffsetAsTimestamp(partition2, 0)));
assertTrue(task.process(0L));
assertThat("task is not idling", !task.timeCurrentIdlingStarted().isPresent());
-
}
@Test
@@ -1881,7 +1919,7 @@ public class StreamTaskTest {
consumer.updateBeginningOffsets(mkMap(mkEntry(repartition, 0L)));
final StreamsConfig config = createConfig();
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -2165,7 +2203,6 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
final long offset = 543L;
- final long consumedOffset = 345L;
when(recordCollector.offsets()).thenReturn(singletonMap(changelogPartition, offset));
when(stateManager.changelogOffsets())
@@ -2478,7 +2515,7 @@ public class StreamTaskTest {
@Test
public void shouldThrowTopologyExceptionIfTaskCreatedForUnknownTopic() {
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
createConfig("100"),
stateManager,
@@ -2545,21 +2582,118 @@ public class StreamTaskTest {
}
@Test
- public void shouldUpdateOffsetIfAllRecordsAreCorrupted() {
+ public void shouldUpdateOffsetIfAllRecordsHaveInvalidTimestamp() {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
- "0",
- LogAndContinueExceptionHandler.class.getName()
- ));
+ task = createStatelessTask(createConfigWithTsExtractor(LogAndSkipOnInvalidTimestamp.class));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ long offset = -1L;
+
+ final List<ConsumerRecord<byte[], byte[]>> records = asList(
+ getConsumerRecordWithInvalidTimestamp(++offset),
+ getConsumerRecordWithInvalidTimestamp(++offset)
+ );
+ consumer.addRecord(records.get(0));
+ consumer.addRecord(records.get(1));
+ task.resumePollingForPartitionsWithAvailableSpace();
+ consumer.poll(Duration.ZERO);
+ task.addRecords(partition1, records);
+ task.updateNextOffsets(partition1, new OffsetAndMetadata(offset + 1, Optional.empty(), ""));
+ task.updateLags();
+
+ assertTrue(task.process(offset));
+ assertTrue(task.commitNeeded());
+ assertThat(
+ task.prepareCommit(),
+ equalTo(mkMap(mkEntry(partition1,
+ new OffsetAndMetadata(offset + 1,
+ new TopicPartitionMetadata(RecordQueue.UNKNOWN, new ProcessorMetadata()).encode()))))
+ );
+ }
+
+ @Test
+ public void shouldUpdateOffsetIfValidRecordFollowsInvalidTimestamp() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ task = createStatelessTask(createConfigWithTsExtractor(LogAndSkipOnInvalidTimestamp.class));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ long offset = -1L;
+
+ final List<ConsumerRecord<byte[], byte[]>> records = asList(
+ getConsumerRecordWithInvalidTimestamp(++offset),
+ getConsumerRecordWithOffsetAsTimestamp(partition1, ++offset)
+ );
+ consumer.addRecord(records.get(0));
+ consumer.addRecord(records.get(1));
+ task.resumePollingForPartitionsWithAvailableSpace();
+ consumer.poll(Duration.ZERO);
+ task.addRecords(partition1, records);
+ task.updateNextOffsets(partition1, new OffsetAndMetadata(offset + 1, Optional.empty(), ""));
+ task.updateLags();
+
+ assertTrue(task.process(offset));
+ assertTrue(task.commitNeeded());
+ assertThat(
+ task.prepareCommit(),
+ equalTo(mkMap(mkEntry(partition1, new OffsetAndMetadata(offset + 1, new TopicPartitionMetadata(offset, new ProcessorMetadata()).encode()))))
+ );
+ }
+
+ @Test
+ public void shouldUpdateOffsetIfInvalidTimestampeRecordFollowsValid() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ task = createStatelessTask(createConfigWithTsExtractor(LogAndSkipOnInvalidTimestamp.class));
task.initializeIfNeeded();
task.completeRestoration(noOpResetter -> { });
long offset = -1;
+
+ final List<ConsumerRecord<byte[], byte[]>> records = asList(
+ getConsumerRecordWithOffsetAsTimestamp(partition1, ++offset),
+ getConsumerRecordWithInvalidTimestamp(++offset));
+ consumer.addRecord(records.get(0));
+ consumer.addRecord(records.get(1));
+ task.resumePollingForPartitionsWithAvailableSpace();
+ consumer.poll(Duration.ZERO);
+ task.addRecords(partition1, records);
+ task.updateNextOffsets(partition1, new OffsetAndMetadata(offset + 1, Optional.empty(), ""));
+
+ task.updateLags();
+
+ assertTrue(task.process(offset));
+ assertTrue(task.commitNeeded());
+ assertThat(
+ task.prepareCommit(),
+ equalTo(mkMap(mkEntry(partition1, new OffsetAndMetadata(1, new TopicPartitionMetadata(0, new ProcessorMetadata()).encode()))))
+ );
+
+ assertTrue(task.process(offset));
+ assertTrue(task.commitNeeded());
+ assertThat(
+ task.prepareCommit(),
+ equalTo(mkMap(mkEntry(partition1, new OffsetAndMetadata(2, new TopicPartitionMetadata(0, new ProcessorMetadata()).encode()))))
+ );
+ }
+
+ @Test
+ public void shouldUpdateOffsetIfAllRecordsAreCorrupted() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ task = createStatelessTask(createConfig(LogAndContinueExceptionHandler.class));
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ long offset = -1L;
+
final List<ConsumerRecord<byte[], byte[]>> records = asList(
getCorruptedConsumerRecordWithOffsetAsTimestamp(++offset),
- getCorruptedConsumerRecordWithOffsetAsTimestamp(++offset));
+ getCorruptedConsumerRecordWithOffsetAsTimestamp(++offset)
+ );
consumer.addRecord(records.get(0));
consumer.addRecord(records.get(1));
task.resumePollingForPartitionsWithAvailableSpace();
@@ -2581,19 +2715,16 @@ public class StreamTaskTest {
public void shouldUpdateOffsetIfValidRecordFollowsCorrupted() {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
- "0",
- LogAndContinueExceptionHandler.class.getName()
- ));
+ task = createStatelessTask(createConfig(LogAndContinueExceptionHandler.class));
task.initializeIfNeeded();
task.completeRestoration(noOpResetter -> { });
- long offset = -1;
+ long offset = -1L;
final List<ConsumerRecord<byte[], byte[]>> records = asList(
getCorruptedConsumerRecordWithOffsetAsTimestamp(++offset),
- getConsumerRecordWithOffsetAsTimestamp(partition1, ++offset));
+ getConsumerRecordWithOffsetAsTimestamp(partition1, ++offset)
+ );
consumer.addRecord(records.get(0));
consumer.addRecord(records.get(1));
task.resumePollingForPartitionsWithAvailableSpace();
@@ -2613,18 +2744,16 @@ public class StreamTaskTest {
public void shouldUpdateOffsetIfCorruptedRecordFollowsValid() {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
- task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
- "0",
- LogAndContinueExceptionHandler.class.getName()));
+ task = createStatelessTask(createConfig(LogAndContinueExceptionHandler.class));
task.initializeIfNeeded();
task.completeRestoration(noOpResetter -> { });
- long offset = -1;
+ long offset = -1L;
final List<ConsumerRecord<byte[], byte[]>> records = asList(
getConsumerRecordWithOffsetAsTimestamp(partition1, ++offset),
- getCorruptedConsumerRecordWithOffsetAsTimestamp(++offset));
+ getCorruptedConsumerRecordWithOffsetAsTimestamp(++offset)
+ );
consumer.addRecord(records.get(0));
consumer.addRecord(records.get(1));
task.resumePollingForPartitionsWithAvailableSpace();
@@ -2676,10 +2805,8 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
- LogAndContinueProcessingExceptionHandler.class.getName()
+ LogAndContinueProcessingExceptionHandler.class
));
final StreamsException streamsException = assertThrows(
@@ -2698,10 +2825,8 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
- LogAndContinueProcessingExceptionHandler.class.getName()
+ LogAndContinueProcessingExceptionHandler.class
));
final Set<TaskId> tasksIds = new HashSet<>();
@@ -2728,10 +2853,8 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
- LogAndContinueProcessingExceptionHandler.class.getName()
+ LogAndContinueProcessingExceptionHandler.class
));
final TaskMigratedException expectedException = new TaskMigratedException("TaskMigratedException", new RuntimeException("Task migrated cause"));
@@ -2751,10 +2874,8 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
- LogAndContinueProcessingExceptionHandler.class.getName()
+ LogAndContinueProcessingExceptionHandler.class
));
task.punctuate(processorStreamTime, 1, PunctuationType.STREAM_TIME, timestamp -> {
@@ -2767,10 +2888,8 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
- LogAndFailProcessingExceptionHandler.class.getName()
+ LogAndFailProcessingExceptionHandler.class
));
final StreamsException streamsException = assertThrows(
@@ -2789,10 +2908,8 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
- NullProcessingExceptionHandler.class.getName()
+ NullProcessingExceptionHandler.class
));
final StreamsException streamsException = assertThrows(
@@ -2812,10 +2929,8 @@ public class StreamTaskTest {
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
- CrashingProcessingExceptionHandler.class.getName()
+ CrashingProcessingExceptionHandler.class
));
final FailedProcessingException streamsException = assertThrows(
@@ -2873,7 +2988,7 @@ public class StreamTaskTest {
singletonList(stateStore),
Collections.singletonMap(storeName, topic1));
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -2915,7 +3030,7 @@ public class StreamTaskTest {
}
};
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -2949,7 +3064,7 @@ public class StreamTaskTest {
emptyMap()
);
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -2988,7 +3103,7 @@ public class StreamTaskTest {
singletonList(stateStore),
logged ? Collections.singletonMap(storeName, storeName + "-changelog") : Collections.emptyMap());
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -3024,7 +3139,7 @@ public class StreamTaskTest {
source1.addChild(processorStreamTime);
source1.addChild(processorSystemTime);
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -3061,7 +3176,7 @@ public class StreamTaskTest {
source1.addChild(processorSystemTime);
source2.addChild(processorSystemTime);
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -3097,7 +3212,7 @@ public class StreamTaskTest {
final StreamsConfig config = createConfig();
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -3130,7 +3245,7 @@ public class StreamTaskTest {
);
final StreamsConfig config = createConfig(eosConfig, "0");
- final InternalProcessorContext context = new ProcessorContextImpl(
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
taskId,
config,
stateManager,
@@ -3207,6 +3322,40 @@ public class StreamTaskTest {
);
}
+ private ConsumerRecord<byte[], byte[]> getConsumerRecordWithInvalidTimestamp(final long offset) {
+ return new ConsumerRecord<>(
+ topic1,
+ 1,
+ offset,
+ -1L, // invalid (negative) timestamp
+ TimestampType.CREATE_TIME,
+ 0,
+ 0,
+ recordKey,
+ recordValue,
+ new RecordHeaders(),
+ Optional.empty()
+ );
+ }
+
+ private ConsumerRecord<byte[], byte[]> getConsumerRecordWithOffsetAsTimestampWithLeaderEpoch(final TopicPartition topicPartition,
+ final long offset,
+ final int leaderEpoch) {
+ return new ConsumerRecord<>(
+ topicPartition.topic(),
+ topicPartition.partition(),
+ offset,
+ offset, // use the offset as the timestamp
+ TimestampType.CREATE_TIME,
+ 0,
+ 0,
+ recordKey,
+ recordValue,
+ new RecordHeaders(),
+ Optional.of(leaderEpoch)
+ );
+ }
+
private ConsumerRecord<byte[], byte[]> getCorruptedConsumerRecordWithOffsetAsTimestamp(final long offset) {
return new ConsumerRecord<>(
topic1,