You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "freeznet (via GitHub)" <gi...@apache.org> on 2026/06/24 12:23:29 UTC
[PR] fix(kinesis): prevent connector crash on checkpoint failure [pulsar-connectors]
freeznet opened a new pull request, #35:
URL: https://github.com/apache/pulsar-connectors/pull/35
(no comment)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(kinesis): add message key mode for Kinesis records [pulsar-connectors]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #35:
URL: https://github.com/apache/pulsar-connectors/pull/35#discussion_r3553866294
##########
kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSourceConfig.java:
##########
@@ -144,6 +144,32 @@ public class KinesisSourceConfig extends BaseKinesisConfig implements Serializab
+ "kinesis.partition.key,kinesis.sequence.number";
private transient Set<String> propertiesToInclude;
+ /**
+ * Determines which value is used as the Pulsar message key for each Kinesis record.
+ */
+ public enum MessageKeyMode {
+ /**
+ * Use the Kinesis record partition key (default; preserves existing downstream partitioning).
+ */
+ PARTITION_KEY,
+ /**
+ * Use the Kinesis shard ID as the message key, routing all records from the same shard to
+ * the same downstream partition. This preserves per-shard ordering and reduces checkpoint
+ * reordering gaps, but may create hot partitions if shard traffic is skewed.
+ */
+ SHARD_ID
+ }
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "PARTITION_KEY",
+ help = "Which value to use as the Pulsar message key for each record. "
+ + "PARTITION_KEY (default) uses the Kinesis record partition key (preserves existing downstream "
+ + "partitioning). SHARD_ID routes all records of a Kinesis shard to the same key, keeping them on "
+ + "one downstream partition to preserve per-shard ordering and reduce checkpoint reordering/gap; "
+ + "note this changes downstream partition distribution and may create hot partitions.")
Review Comment:
The config help text uses "reordering/gap" which reads like a typo and is a bit unclear. Consider changing this to "reordering gaps".
##########
kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisRecordProcessor.java:
##########
@@ -78,34 +94,74 @@ private void tryCheckpointWithRetry(RecordProcessorCheckpointer checkpointer,
checkpointer.checkpoint(checkpoint.sequenceNumber(), checkpoint.subSequenceNumber());
lastCheckpointSequenceNumber = checkpoint;
log.info("Successfully checkpointed shard {} at {}", kinesisShardId, checkpoint);
- isCheckpointing.set(false);
- checkpointExecutor.schedule(this::triggerCheckpoint, checkpointInterval, TimeUnit.MILLISECONDS);
+ scheduleNextCheckpoint();
} catch (ThrottlingException | KinesisClientLibDependencyException e) {
if (attempt >= numRetries) {
- log.error("Checkpoint for shard {} failed after {} attempts at {}. Terminating.",
- kinesisShardId, numRetries, checkpoint, e);
- sourceContext.fatal(e);
+ // A failed periodic checkpoint is recoverable: skip this round instead of terminating the
+ // connector, and let the next scheduled checkpoint retry with the latest acked position.
+ log.error("Checkpoint for shard {} failed after {} attempts at {}. Skipping this round; "
+ + "will retry on the next interval.", kinesisShardId, numRetries, checkpoint, e);
+ scheduleNextCheckpoint();
} else {
log.warn("Throttling/Dependency error on checkpoint for shard {} at {}. Scheduling retry {} "
+ "after {}ms.", kinesisShardId, checkpoint, attempt + 1, backoffTime);
- checkpointExecutor.schedule(() -> tryCheckpointWithRetry(checkpointer, checkpoint, attempt + 1),
- backoffTime, TimeUnit.MILLISECONDS);
+ safeSchedule(() -> tryCheckpointWithRetry(checkpointer, checkpoint, attempt + 1), backoffTime);
}
+ } catch (IllegalArgumentException e) {
+ // KCL rejects a backward / out-of-range checkpoint with IllegalArgumentException. With the
+ // contiguous-prefix logic this should no longer happen, so surface it as a metric for alerting,
+ // but still skip-and-reschedule rather than crashing the connector.
+ log.warn("KCL rejected a backward/out-of-range checkpoint for shard {} at {}. Skipping this round; "
+ + "will retry on the next interval.", kinesisShardId, checkpoint, e);
+ sourceContext.recordMetric("kinesisBackwardCheckpointDetected", 1);
+ scheduleNextCheckpoint();
} catch (Exception e) {
- log.error("Caught a non-retryable exception for shard {} during checkpoint at {}. Terminating.",
- kinesisShardId, checkpoint, e);
- sourceContext.fatal(e);
+ // Other non-retryable errors must NOT crash the connector. Skip this round; a later ack with a
+ // higher position will advance the checkpoint on a subsequent cycle.
+ log.error("Non-retryable exception during checkpoint for shard {} at {}. Skipping this round; "
+ + "will retry on the next interval.", kinesisShardId, checkpoint, e);
+ scheduleNextCheckpoint();
+ }
+ }
+
+ private void scheduleNextCheckpoint() {
+ isCheckpointing.set(false);
+ safeSchedule(this::triggerCheckpoint, checkpointInterval);
+ }
+
+ private void safeSchedule(Runnable task, long delayMillis) {
+ try {
+ checkpointExecutor.schedule(task, delayMillis, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // The checkpoint executor has been shut down (connector is closing); nothing left to schedule.
+ log.debug("Checkpoint executor rejected a task for shard {} (likely shutting down).", kinesisShardId);
}
}
public void updateSequenceNumberToCheckpoint(String sequenceNumber, long subSequenceNumber) {
- CheckpointSequenceNumber newCheckpoint = new CheckpointSequenceNumber(sequenceNumber, subSequenceNumber);
- log.debug("{} Updating sequence number to checkpoint {}", kinesisShardId, newCheckpoint);
- this.sequenceNumberNeedToCheckpoint = newCheckpoint;
+ CheckpointSequenceNumber acked = new CheckpointSequenceNumber(sequenceNumber, subSequenceNumber);
+ log.debug("{} Record acked at {}", kinesisShardId, acked);
+ synchronized (checkpointLock) {
+ ackedAwaitingCollapse.add(acked);
+ // Collapse the longest contiguous, fully-acked prefix (in delivery order) into highestContiguousAcked.
+ // Stop at the first un-acked head: we must never advance the checkpoint past a still-in-flight
+ // lower sequence, otherwise a crash would resume AFTER it and lose it.
+ while (!deliveredInOrder.isEmpty()) {
+ CheckpointSequenceNumber head = deliveredInOrder.peekFirst();
+ if (!ackedAwaitingCollapse.remove(head)) {
+ break;
+ }
+ deliveredInOrder.pollFirst();
+ highestContiguousAcked = head;
+ }
+ }
this.numRecordsInFlight.decrementAndGet();
}
public void failed() {
+ // Intentionally do NOT remove this record's position from deliveredInOrder: a record that failed to
+ // be delivered downstream must keep blocking the contiguous-prefix collapse so the checkpoint can
+ // never advance past it. fatal() terminates the connector, which reprocesses from the last checkpoint.
numRecordsInFlight.decrementAndGet();
sourceContext.fatal(new PulsarClientException("Failed to process Kinesis records due send to pulsar topic"));
Review Comment:
The exception message has a grammatical error ("due send") and uses inconsistent capitalization for Pulsar, which makes logs harder to read/search. Consider rewording to a clear failure description.
##########
kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisRecordProcessorTest.java:
##########
@@ -93,32 +95,105 @@ public void testScheduledCheckpointAfterAck() throws Exception {
}
@Test
- public void testCheckpointLogicOnlyAdvancesForward() throws Exception {
- // Arrange
+ public void testOutOfOrderAcksDoNotAdvancePastGap() throws Exception {
+ // Arrange: deliver three records of the same shard IN ORDER via processRecords. KCL delivery order
+ // is authoritative, so the contiguous prefix is seq-1/0 -> seq-1/1 -> seq-1/2.
recordProcessor.initialize(createMockInitializationInput());
verify(checkpointExecutor, Mockito.times(1))
.schedule(scheduledTaskCaptor.capture(), anyLong(), any(TimeUnit.class));
Runnable scheduledCheckpointTask = scheduledTaskCaptor.getValue();
- ProcessRecordsInput processRecordsInput = createMockProcessRecordsInput(checkpointer);
- recordProcessor.processRecords(processRecordsInput);
+ recordProcessor.processRecords(createMockProcessRecordsInput(
+ createMockKinesisRecord("seq-1", 0L),
+ createMockKinesisRecord("seq-1", 1L),
+ createMockKinesisRecord("seq-1", 2L)));
- // Act & Assert 1: Ack a later sub-sequence number first.
- recordProcessor.updateSequenceNumberToCheckpoint("seq-1", 5L);
+ // Act & Assert 1: ack ONLY the highest (seq-1/2). There is still a gap at seq-1/0, so the
+ // contiguous prefix is empty and nothing must be checkpointed.
+ recordProcessor.updateSequenceNumberToCheckpoint("seq-1", 2L);
+ scheduledCheckpointTask.run();
+ verify(checkpointer, never()).checkpoint(any(String.class), anyLong());
+
+ // Act & Assert 2: ack the lowest (seq-1/0). The contiguous prefix now covers exactly seq-1/0,
+ // because seq-1/1 is still un-acked. Checkpoint must be seq-1/0, NOT the already-acked seq-1/2.
+ recordProcessor.updateSequenceNumberToCheckpoint("seq-1", 0L);
+ scheduledCheckpointTask.run();
+ verify(checkpointer, times(1)).checkpoint("seq-1", 0L);
+
+ // Act & Assert 3: ack the middle (seq-1/1). The prefix now collapses through seq-1/1 and the
+ // already-acked seq-1/2, advancing the checkpoint to seq-1/2.
+ recordProcessor.updateSequenceNumberToCheckpoint("seq-1", 1L);
+ scheduledCheckpointTask.run();
+ verify(checkpointer, times(1)).checkpoint("seq-1", 2L);
+ }
+
+ @Test
+ public void testIncidentScenarioOutOfOrderAcksAcrossSequences() throws Exception {
+ // Mirrors INCIDENT 3800: distinct sequence numbers delivered in order seq-100, seq-101, seq-102.
+ recordProcessor.initialize(createMockInitializationInput());
Review Comment:
This test comment references an "INCIDENT 3800" without any public context or link. In an open-source test suite, prefer a self-contained description (or link to a public issue) so future readers can understand why the scenario matters.
##########
kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisRecordProcessor.java:
##########
@@ -50,13 +53,25 @@ private record CheckpointSequenceNumber(String sequenceNumber, long subSequenceN
private final LinkedBlockingQueue<KinesisRecord> queue;
private final SourceContext sourceContext;
private final Set<String> propertiesToInclude;
+ private final KinesisSourceConfig.MessageKeyMode messageKeyMode;
private final ScheduledExecutorService checkpointExecutor;
private final AtomicReference<RecordProcessorCheckpointer> checkpointerRef = new AtomicReference<>();
private final AtomicBoolean isCheckpointing = new AtomicBoolean(false);
private String kinesisShardId;
private final AtomicInteger numRecordsInFlight = new AtomicInteger(0);
- private volatile CheckpointSequenceNumber sequenceNumberNeedToCheckpoint = null;
+ // All checkpoint-position state below is guarded by checkpointLock. Acks arrive concurrently on the
+ // downstream sink's send-completion callback threads, processRecords runs on the KCL thread, and
+ // triggerCheckpoint runs on the checkpointExecutor thread, so the deque/set/highest-acked triple must
+ // be mutated atomically as a group.
+ private final Object checkpointLock = new Object();
+ // Positions appended in processRecords, in KCL delivery (sequence) order.
+ private final ArrayDeque<CheckpointSequenceNumber> deliveredInOrder = new ArrayDeque<>();
+ // Positions acked but not yet collapsed into the contiguous prefix (acked ahead of an earlier gap).
+ private final HashSet<CheckpointSequenceNumber> ackedAwaitingCollapse = new HashSet<>();
+ // The highest position whose entire delivery-order prefix has been acked. The ONLY thing we checkpoint.
+ private CheckpointSequenceNumber highestContiguousAcked = null;
+
private volatile CheckpointSequenceNumber lastCheckpointSequenceNumber = null;
Review Comment:
The PR title indicates this is only adding a Kinesis message-key mode, but this change set also materially alters checkpointing behavior (contiguous-prefix tracking, retry/termination semantics, and SHARD_END gating). If intentional, please update the PR title/description accordingly or split the checkpointing changes into a separate PR to keep review scope clear.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(kinesis): add message key mode for Kinesis records [pulsar-connectors]
Posted by "freeznet (via GitHub)" <gi...@apache.org>.
freeznet merged PR #35:
URL: https://github.com/apache/pulsar-connectors/pull/35
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org