Fix how failures are handled for shard partitions in DDB source

[graytaylor0]

Description

This change fixes how the DynamoDB source handles failures with acknowledgments, including ack timeouts, negative acknowledgments, and failure to update partition state.

When these failure cases happen, we will now signal to the shardConsumer to stop processing the shard.

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[dlvenable]

Please make this package protected.

Also, update the constructor above to use this.

public ShardAcknowledgementManager(final AcknowledgementSetManager acknowledgementSetManager,
                                       final EnhancedSourceCoordinator sourceCoordinator,
                                       final DynamoDBSourceConfig dynamoDBSourceConfig,
                                       final Consumer<StreamPartition> stopWorkerConsumer) {
  this(acknowledgementSetManager, sourceCoordinator, dynamoDBSourceConfig, createExecutorService(stopWorkConsumer));
}

private static ExecutorService createExecutorService(final Consumer<StreamPartition> stopWorkerConsumer) {
  ExecutorService executorService = Executors.newSingleThreadExecutor(BackgroundThreadFactory.defaultExecutorThreadFactory("dynamodb-shard-ack-monitor"));
  executorService.submit(() -> monitorAcknowledgements(stopWorkConsumer));
  return executorService.
}

[graytaylor0]

private static ExecutorService createExecutorService(final Consumer<StreamPartition> stopWorkerConsumer) {
  ExecutorService executorService = Executors.newSingleThreadExecutor(BackgroundThreadFactory.defaultExecutorThreadFactory("dynamodb-shard-ack-monitor"));
  executorService.submit(() -> monitorAcknowledgements(stopWorkConsumer));
  return executorService.
}

monitorAcknowledgments method can't be used in a static method

[dlvenable]

That is just a Runnable. So you should be able to add a Runnable as a parameter to the static method.

Then:

this(acknowledgementSetManager, sourceCoordinator, dynamoDBSourceConfig, createExecutorService(stopWorkConsumer), () -> monitorAcknowledgements(stopWorkConsumer))

[dlvenable]

Please remove this commented out code.

[dlvenable]

Maybe for a future PR, but this and the exception seem to just result in ShardConsumer threads just going away without any real tracking of them.

[graytaylor0]

What would the ideal behavior be here? This does just have the ShardConsumer exit

[dlvenable]

I think we may be ok here. I found this code that handles exceptions:

data-prepper/data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/StreamScheduler.java

Lines 157 to 164 in 2c66d59

	} else {
	LOG.error("Received an exception while processing shard {}, giving up shard: {}", streamPartition.getShardId(), ex);
	if (dynamoDBSourceConfig.isAcknowledgmentsEnabled()) {
	shardAcknowledgementManager.giveUpPartition(streamPartition);
	} else {
	coordinator.giveUpPartition(streamPartition);
	}
	}

[dlvenable]

Is giveUpPartition idempotent? If not, we might need a change here since we have already lost the partition in this particular case.

[graytaylor0]

Yes this is what happens when ShardConsumer throws an error.

And giveUpPartition would work if called twice as long as another worker hasn't already grabbed ownership of the shard. However, I tried to handle this to only be called once by putting this check in the giveUpPartition method of shardAcknowledgmentManager

public void giveUpPartition(final StreamPartition streamPartition) {
        if (isStillTrackingShard(streamPartition)) {
            LOG.debug("Adding partition {} to give up list", streamPartition.getPartitionKey());
            partitionsToGiveUp.add(streamPartition);
            partitionsToRemove.add(streamPartition);
        }
    }

So basically this call will only mark the partition to be given up by the acknowledgment manager and isn't actually calling give up partition. So it will only be called once.

[san81]

it looks like all the collections the logic is dealing with - are already synchronized. Do we still need this lock?

[dlvenable]

Any given collection is synchronized. But, this block of code is modifying multiple collections together.

[dlvenable]

Improve this log message:

LOG.warn("Not creating acknowledgment set since shard is not tracked: {}", e.getMessage());

There are two reasons for this:

• When somebody sees this warning it will be hard to know where is originated unless line logging is unabled.
• Unknown strings shouldn't be passed to a logger since it is treated as a format string.

[graytaylor0]

Makes sense will change this