This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 306830fcd02 NIFI-16066 Release lingering rename in ConsumeKinesis
(#11385)
306830fcd02 is described below
commit 306830fcd0220ae994b3c1f7f2c5e42aacfcc1fc
Author: Alaksiej Ščarbaty <[email protected]>
AuthorDate: Thu Jul 2 17:03:31 2026 +0200
NIFI-16066 Release lingering rename in ConsumeKinesis (#11385)
* NIFI-16066 Release lingering rename lock when error occurs in
ConsumeKinesis
When an error (e.g. a failed checkpoint copy) occurred while a node held
the rename lock in LegacyCheckpointMigrator, the renameOwner marker was
left on the migration table, blocking all future rename attempts until
the stale-lock timeout elapsed. Release the rename lock on error, on all
three lock-holding paths, so the rename can be retried. The release is
owner-guarded (renameOwner = :owner) so a slow-failing node cannot wipe a
lock another node has already force-acquired via the stale-lock takeover,
and it is best-effort so a transient DynamoDB error while releasing does
not mask the original failure.
Also make waitForTableRenamed treat a rename as complete only once the
migration table has been dropped, not merely when the checkpoint table
has the new schema. The new checkpoint table is created empty before the
copy runs, so schema presence alone does not indicate the migration
finished.
---
.../aws/kinesis/LegacyCheckpointMigrator.java | 81 +++++++++++++-----
.../aws/kinesis/KinesisShardManagerTest.java | 95 ++++++++++++++++++++++
2 files changed, 156 insertions(+), 20 deletions(-)
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 3118b70d1a0..e42e6c7023d 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
@@ -85,9 +85,14 @@ final class LegacyCheckpointMigrator {
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);
+ try {
+ CheckpointTableUtils.waitForTableActive(dynamoDbClient,
logger, lingeringMigration);
+ CheckpointTableUtils.copyCheckpointItems(dynamoDbClient,
logger, lingeringMigration, checkpointTableName);
+ CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
lingeringMigration);
+ } catch (final Exception e) {
+ clearRenameLock(lingeringMigration);
+ throw e;
+ }
} else {
waitForTableRenamed(lingeringMigration);
}
@@ -114,12 +119,17 @@ final class LegacyCheckpointMigrator {
void renameMigrationTable(final String migrationTableName) {
if (acquireRenameLock(migrationTableName)) {
- CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.waitForTableDeleted(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.createNewSchemaTable(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.waitForTableActive(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.copyCheckpointItems(dynamoDbClient, logger,
migrationTableName, checkpointTableName);
- CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
migrationTableName);
+ try {
+ CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
checkpointTableName);
+ CheckpointTableUtils.waitForTableDeleted(dynamoDbClient,
logger, checkpointTableName);
+ CheckpointTableUtils.createNewSchemaTable(dynamoDbClient,
logger, checkpointTableName);
+ CheckpointTableUtils.waitForTableActive(dynamoDbClient,
logger, checkpointTableName);
+ CheckpointTableUtils.copyCheckpointItems(dynamoDbClient,
logger, migrationTableName, checkpointTableName);
+ CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
migrationTableName);
+ } catch (final Exception e) {
+ clearRenameLock(migrationTableName);
+ throw e;
+ }
} else {
waitForTableRenamed(migrationTableName);
}
@@ -181,11 +191,38 @@ final class LegacyCheckpointMigrator {
"shardId",
AttributeValue.builder().s(CheckpointTableUtils.MIGRATION_MARKER_SHARD_ID).build());
}
+ private void clearRenameLock(final String migrationTableName) {
+ try {
+ final UpdateItemRequest request = UpdateItemRequest.builder()
+ .tableName(migrationTableName)
+ .key(migrationMarkerKey())
+ .updateExpression("REMOVE renameOwner, renameStartedAt")
+ .conditionExpression("renameOwner = :owner")
+ .expressionAttributeValues(Map.of(
+ ":owner",
AttributeValue.builder().s(nodeId).build()))
+ .build();
+ dynamoDbClient.updateItem(request);
+ logger.info("Released rename lock on migration table [{}] after an
error", migrationTableName);
+ } catch (final ConditionalCheckFailedException e) {
+ logger.debug("Rename lock on migration table [{}] is no longer
held by this node; leaving it in place", migrationTableName);
+ } catch (final ResourceNotFoundException e) {
+ logger.debug("Migration table [{}] no longer exists; rename lock
already released", migrationTableName);
+ } catch (final Exception e) {
+ logger.warn("Failed to release rename lock on migration table [{}]
after an error; it will expire via the stale-lock timeout",
+ migrationTableName, e);
+ }
+ }
+
+ private boolean isRenameCompleted(final String migrationTableName) {
+ return CheckpointTableUtils.getTableSchema(dynamoDbClient,
checkpointTableName) == CheckpointTableUtils.TableSchema.NEW
+ && CheckpointTableUtils.getTableSchema(dynamoDbClient,
migrationTableName) == CheckpointTableUtils.TableSchema.NOT_FOUND;
+ }
+
private void waitForTableRenamed(final String migrationTableName) {
for (int i = 0; i < RENAME_POLL_MAX_ATTEMPTS; i++) {
- if (CheckpointTableUtils.getTableSchema(dynamoDbClient,
checkpointTableName)
- == CheckpointTableUtils.TableSchema.NEW) {
- logger.info("Migration table rename complete; table [{}] is
now available", checkpointTableName);
+ if (isRenameCompleted(migrationTableName)) {
+ logger.info("Migration table rename complete; table [{}] is
now available and migration table [{}] has been dropped",
+ checkpointTableName, migrationTableName);
return;
}
@@ -199,14 +236,18 @@ final class LegacyCheckpointMigrator {
logger.warn("Timed out waiting for migration table rename; attempting
stale lock takeover");
if (forceAcquireStaleRenameLock(migrationTableName)) {
- CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.waitForTableDeleted(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.createNewSchemaTable(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.waitForTableActive(dynamoDbClient, logger,
checkpointTableName);
- CheckpointTableUtils.copyCheckpointItems(dynamoDbClient, logger,
migrationTableName, checkpointTableName);
- CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
migrationTableName);
- } else if (CheckpointTableUtils.getTableSchema(dynamoDbClient,
checkpointTableName)
- == CheckpointTableUtils.TableSchema.NEW) {
+ try {
+ CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
checkpointTableName);
+ CheckpointTableUtils.waitForTableDeleted(dynamoDbClient,
logger, checkpointTableName);
+ CheckpointTableUtils.createNewSchemaTable(dynamoDbClient,
logger, checkpointTableName);
+ CheckpointTableUtils.waitForTableActive(dynamoDbClient,
logger, checkpointTableName);
+ CheckpointTableUtils.copyCheckpointItems(dynamoDbClient,
logger, migrationTableName, checkpointTableName);
+ CheckpointTableUtils.deleteTable(dynamoDbClient, logger,
migrationTableName);
+ } catch (final Exception e) {
+ clearRenameLock(migrationTableName);
+ throw e;
+ }
+ } else if (isRenameCompleted(migrationTableName)) {
logger.info("Migration table rename completed during takeover
attempt");
} else {
throw new ProcessException(
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 c76ba094e3d..57cd94cb2eb 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
@@ -51,9 +51,12 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -362,6 +365,98 @@ class KinesisShardManagerTest {
"Migration table must be deleted only after the recovery copy
succeeds");
}
+ /**
+ * When an error occurs while a node holds the rename lock (for example
the checkpoint copy
+ * fails after the new checkpoint table has been recreated), the rename
lock on the migration
+ * table is released so a later attempt can re-acquire it and retry, and
the migration table
+ * is left in place so no checkpoints are lost.
+ */
+ @Test
+ void testRenameMigrationTableReleasesRenameLockOnCopyError() {
+ final AtomicInteger mainTableDescribeCount = new AtomicInteger();
+
when(dynamoDb.describeTable(any(DescribeTableRequest.class))).thenAnswer(invocation
-> {
+ final DescribeTableRequest req = invocation.getArgument(0);
+ if ("test-table".equals(req.tableName())) {
+ if (mainTableDescribeCount.incrementAndGet() <= 3) {
+ throw ResourceNotFoundException.builder().build();
+ }
+ return newSchemaActiveResponse();
+ }
+ if ("test-table_migration".equals(req.tableName())) {
+ return newSchemaActiveResponse();
+ }
+ throw ResourceNotFoundException.builder().build();
+ });
+
+
when(dynamoDb.updateItem(any(UpdateItemRequest.class))).thenReturn(UpdateItemResponse.builder().build());
+
when(dynamoDb.createTable(any(CreateTableRequest.class))).thenReturn(CreateTableResponse.builder().build());
+
when(dynamoDb.deleteTable(any(DeleteTableRequest.class))).thenReturn(DeleteTableResponse.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);
+
+ assertThrows(InternalServerErrorException.class, () ->
manager.ensureCheckpointTableExists(),
+ "A failed checkpoint copy under the rename lock must
propagate");
+
+ final ArgumentCaptor<UpdateItemRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateItemRequest.class);
+ verify(dynamoDb, atLeastOnce()).updateItem(updateCaptor.capture());
+ final boolean lockReleased = updateCaptor.getAllValues().stream()
+ .anyMatch(req -> "test-table_migration".equals(req.tableName())
+ && req.updateExpression() != null
+ && req.updateExpression().startsWith("REMOVE
renameOwner"));
+ assertTrue(lockReleased, "Rename lock (renameOwner) must be released
on the migration table when the rename fails");
+
+ final ArgumentCaptor<DeleteTableRequest> deleteCaptor =
ArgumentCaptor.forClass(DeleteTableRequest.class);
+ verify(dynamoDb, times(1)).deleteTable(deleteCaptor.capture());
+ final boolean migrationDeleted = deleteCaptor.getAllValues().stream()
+ .map(DeleteTableRequest::tableName)
+ .anyMatch("test-table_migration"::equals);
+ assertFalse(migrationDeleted, "Migration table must not be deleted
when the rename copy fails");
+ }
+
+ /**
+ * A node that does not hold the rename lock treats the rename as complete
only once the
+ * migration table has been dropped. Because the new checkpoint table is
created empty before
+ * the checkpoint copy runs, the presence of the new schema on the
checkpoint table alone is
+ * not sufficient to conclude the migration finished.
+ */
+ @Test
+ void testWaitForTableRenamedRequiresMigrationTableDropped() {
+ final AtomicInteger migrationDescribeCount = new AtomicInteger();
+
when(dynamoDb.describeTable(any(DescribeTableRequest.class))).thenAnswer(invocation
-> {
+ final DescribeTableRequest req = invocation.getArgument(0);
+ if ("test-table".equals(req.tableName())) {
+ return newSchemaActiveResponse();
+ }
+ if ("test-table_migration".equals(req.tableName())) {
+ // Count 1: findMigrationTable() in
completeLingeringMigration; count 2+: waitForTableRenamed poll loop.
+ if (migrationDescribeCount.incrementAndGet() == 1) {
+ return newSchemaActiveResponse();
+ }
+ throw ResourceNotFoundException.builder().build();
+ }
+ throw ResourceNotFoundException.builder().build();
+ });
+
+ // Losing the rename-lock race: acquireRenameLock's conditional update
fails.
+ when(dynamoDb.updateItem(any(UpdateItemRequest.class)))
+ .thenThrow(ConditionalCheckFailedException.builder().build());
+
+ manager.ensureCheckpointTableExists();
+
+ assertTrue(migrationDescribeCount.get() >= 2,
+ "waitForTableRenamed must confirm the migration table was
dropped, not just that the checkpoint table has the new schema");
+ verify(dynamoDb, never()).deleteTable(any(DeleteTableRequest.class));
+ }
+
private static DescribeTableResponse newSchemaActiveResponse() {
final KeySchemaElement hashKey = KeySchemaElement.builder()
.attributeName("streamName")
