This is an automated email from the ASF dual-hosted git repository.
markap14 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 6f1722052b5 NIFI-16003 Re-copy checkpoints before deleting lingering
migration table (#11320)
6f1722052b5 is described below
commit 6f1722052b56c6ad6f932fcfa5d547b1848e0d9b
Author: Alaksiej Ščarbaty <[email protected]>
AuthorDate: Wed Jun 10 18:23:25 2026 +0200
NIFI-16003 Re-copy checkpoints before deleting lingering migration table
(#11320)
When a prior legacy-checkpoint migration failed mid-flight (e.g. step 5
copy threw because DynamoDB's request router still cached the old table
schema after recreation), the migration table was left behind as the
only surviving copy of the per-shard checkpoint state. On restart,
LegacyCheckpointMigrator.cleanupLingeringMigration would unconditionally
delete that table, silently losing all checkpoints and forcing every
shard back to INITIAL_STREAM_POSITION.
Re-run the copy from the migration table into the checkpoint table
before deleting it. This is safe because any failure in the rename
sequence propagates out of ensureCheckpointTableExists and prevents the
processor from starting, so the checkpoint table cannot have advanced
past the migration table's contents — the re-copy either restores lost
state or re-puts identical rows.
---
.../aws/kinesis/KinesisShardManager.java | 2 +-
.../aws/kinesis/LegacyCheckpointMigrator.java | 13 +++++--
.../aws/kinesis/KinesisShardManagerTest.java | 45 ++++++++++++++++++----
3 files changed, 49 insertions(+), 11 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManager.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManager.java
index 53fbfd41bd3..cfaf218ae5d 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManager.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManager.java
@@ -149,7 +149,7 @@ final class KinesisShardManager {
}
case NEW -> {
CheckpointTableUtils.waitForTableActive(dynamoDbClient,
logger, checkpointTableName);
- migrator.cleanupLingeringMigration();
+ migrator.completeLingeringMigration();
}
case LEGACY -> {
migrator.migrateAndRename();
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/LegacyCheckpointMigrator.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/LegacyCheckpointMigrator.java
index d994094d2c4..3118b70d1a0 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/LegacyCheckpointMigrator.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/LegacyCheckpointMigrator.java
@@ -77,13 +77,20 @@ final class LegacyCheckpointMigrator {
return null;
}
- void cleanupLingeringMigration() {
+ void completeLingeringMigration() {
final String lingeringMigration = findMigrationTable();
if (lingeringMigration == null) {
return;
}
- logger.info("Deleting orphaned migration table [{}]; legacy checkpoint
table [{}] retains original data", lingeringMigration, checkpointTableName);
- CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
lingeringMigration);
+ logger.info("Checkpoint table [{}] is in place but migration table
[{}] still exists from a prior rename; not all checkpoints may have been
migrated, re-copying before deletion",
+ checkpointTableName, lingeringMigration);
+ if (acquireRenameLock(lingeringMigration)) {
+ CheckpointTableUtils.waitForTableActive(dynamoDbClient, logger,
lingeringMigration);
+ CheckpointTableUtils.copyCheckpointItems(dynamoDbClient, logger,
lingeringMigration, checkpointTableName);
+ CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
lingeringMigration);
+ } else {
+ waitForTableRenamed(lingeringMigration);
+ }
}
void migrateAndRename() {
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManagerTest.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManagerTest.java
index 31c900d701f..c76ba094e3d 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManagerTest.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisShardManagerTest.java
@@ -31,6 +31,7 @@ import
software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest;
import software.amazon.awssdk.services.dynamodb.model.DescribeTableResponse;
import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
import software.amazon.awssdk.services.dynamodb.model.GetItemResponse;
+import
software.amazon.awssdk.services.dynamodb.model.InternalServerErrorException;
import software.amazon.awssdk.services.dynamodb.model.KeySchemaElement;
import software.amazon.awssdk.services.dynamodb.model.KeyType;
import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
@@ -51,6 +52,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -306,12 +308,17 @@ class KinesisShardManagerTest {
}
/**
- * Verifies that when the configured table has the new schema but a
lingering migration
- * table exists (crash after copy but before migration table deletion),
the items are
- * copied and the migration table is cleaned up.
+ * Regression test for NIFI-16003: cleanup of a lingering migration table
must NOT delete
+ * the migration table unless the recovery copy succeeds. Previously it
was deleted
+ * unconditionally, causing checkpoint loss when the prior copy had failed
(e.g. due to
+ * a stale DynamoDB router schema cache after table recreation).
+ *
+ * Phase 1 forces the copy to fail and verifies the migration table is
preserved.
+ * Phase 2 retries with a working copy and verifies items are recopied and
the migration
+ * table is deleted.
*/
@Test
- void testEnsureCheckpointTableCleansUpLingeringMigrationTable() {
+ void testEnsureCheckpointTableRecoversCopyFromLingeringMigrationTable() {
when(dynamoDb.describeTable(any(DescribeTableRequest.class))).thenAnswer(invocation
-> {
final DescribeTableRequest req = invocation.getArgument(0);
if ("test-table".equals(req.tableName()) ||
"test-table_migration".equals(req.tableName())) {
@@ -321,14 +328,38 @@ class KinesisShardManagerTest {
});
when(dynamoDb.deleteTable(any(DeleteTableRequest.class))).thenReturn(DeleteTableResponse.builder().build());
-
when(dynamoDb.scan(any(ScanRequest.class))).thenReturn(ScanResponse.builder().build());
+
when(dynamoDb.updateItem(any(UpdateItemRequest.class))).thenReturn(UpdateItemResponse.builder().build());
+
+ final Map<String, AttributeValue> checkpointItem = Map.of(
+ "streamName",
AttributeValue.builder().s("test-stream").build(),
+ "shardId", AttributeValue.builder().s("shard-1").build(),
+ "sequenceNumber", AttributeValue.builder().s("12345").build());
+
when(dynamoDb.scan(any(ScanRequest.class))).thenReturn(ScanResponse.builder().items(checkpointItem).build());
+
+ final InternalServerErrorException copyFailure =
InternalServerErrorException.builder()
+ .message("simulated transient DynamoDB failure during
checkpoint copy")
+ .build();
+ when(dynamoDb.putItem(any(PutItemRequest.class)))
+ .thenThrow(copyFailure)
+ .thenReturn(PutItemResponse.builder().build());
+
+ assertThrows(InternalServerErrorException.class, () ->
manager.ensureCheckpointTableExists(),
+ "A failed recovery copy must propagate so the processor does
not start with missing checkpoints");
+ verify(dynamoDb, never()).deleteTable(any(DeleteTableRequest.class));
manager.ensureCheckpointTableExists();
- verify(dynamoDb, never()).createTable(any(CreateTableRequest.class));
+ final ArgumentCaptor<PutItemRequest> putCaptor =
ArgumentCaptor.forClass(PutItemRequest.class);
+ verify(dynamoDb, times(2)).putItem(putCaptor.capture());
+ for (final PutItemRequest req : putCaptor.getAllValues()) {
+ assertEquals("test-table", req.tableName(), "Recovery copies must
target the configured checkpoint table");
+ assertEquals("12345", req.item().get("sequenceNumber").s(),
"Checkpoint data must be preserved across the failed copy");
+ }
+
final ArgumentCaptor<DeleteTableRequest> deleteCaptor =
ArgumentCaptor.forClass(DeleteTableRequest.class);
verify(dynamoDb).deleteTable(deleteCaptor.capture());
- assertEquals("test-table_migration",
deleteCaptor.getValue().tableName(), "Lingering migration table should be
deleted");
+ assertEquals("test-table_migration",
deleteCaptor.getValue().tableName(),
+ "Migration table must be deleted only after the recovery copy
succeeds");
}
private static DescribeTableResponse newSchemaActiveResponse() {
