class DynamoDBLockProvider implements LockProvider {
    static final String LOCK_UNTIL = "lockUntil";
    static final String LOCKED_AT = "lockedAt";
    static final String LOCKED_BY = "lockedBy";
    static final String ID = "_id";

    private static final String OBTAIN_LOCK_QUERY =
            "set " + LOCK_UNTIL + " = :lockUntil, " + LOCKED_AT + " = :lockedAt, " + LOCKED_BY + " = :lockedBy";
    private static final String OBTAIN_LOCK_CONDITION =
            LOCK_UNTIL + " <= :lockedAt or attribute_not_exists(" + LOCK_UNTIL + ")";
    private static final String RELEASE_LOCK_QUERY = "set " + LOCK_UNTIL + " = :lockUntil";

    private final String hostname;
    private final DynamoDbClient dynamoDbClient;
    private final String tableName;

    /**
     * Uses DynamoDB to coordinate locks
     *
     * @param dynamoDbClient
     *            v2 of DynamoDB client
     * @param tableName
     *            the lock table name
     */
    public DynamoDBLockProvider(@NonNull DynamoDbClient dynamoDbClient, @NonNull String tableName) {
        this.dynamoDbClient = requireNonNull(dynamoDbClient, "dynamoDbClient can not be null");
        this.tableName = requireNonNull(tableName, "tableName can not be null");
        this.hostname = Utils.getHostname();
    }

    @Override
    @NonNull
    public Optional<SimpleLock> lock(@NonNull LockConfiguration lockConfiguration) {<FILL_FUNCTION_BODY>}

    private static AttributeValue attr(String lockUntilIso) {
        return AttributeValue.builder().s(lockUntilIso).build();
    }

    private Instant now() {
        return ClockProvider.now();
    }

    private static final class DynamoDBLock extends AbstractSimpleLock {
        private final DynamoDbClient dynamoDbClient;
        private final String tableName;

        private DynamoDBLock(DynamoDbClient dynamoDbClient, String tableName, LockConfiguration lockConfiguration) {
            super(lockConfiguration);
            this.dynamoDbClient = dynamoDbClient;
            this.tableName = tableName;
        }

        @Override
        public void doUnlock() {
            // Set lockUntil to now or lockAtLeastUntil whichever is later
            String unlockTimeIso = toIsoString(lockConfiguration.getUnlockTime());

            Map<String, AttributeValue> key = singletonMap(ID, attr(lockConfiguration.getName()));

            Map<String, AttributeValue> attributeUpdates = singletonMap(":lockUntil", attr(unlockTimeIso));

            UpdateItemRequest request = UpdateItemRequest.builder()
                    .tableName(tableName)
                    .key(key)
                    .updateExpression(RELEASE_LOCK_QUERY)
                    .expressionAttributeValues(attributeUpdates)
                    .returnValues(ReturnValue.UPDATED_NEW)
                    .build();

            dynamoDbClient.updateItem(request);
        }
    }
}

        String nowIso = toIsoString(now());
        String lockUntilIso = toIsoString(lockConfiguration.getLockAtMostUntil());

        Map<String, AttributeValue> key = singletonMap(ID, attr(lockConfiguration.getName()));

        Map<String, AttributeValue> attributeUpdates =
                Map.of(":lockUntil", attr(lockUntilIso), ":lockedAt", attr(nowIso), ":lockedBy", attr(hostname));

        UpdateItemRequest request = UpdateItemRequest.builder()
                .tableName(tableName)
                .key(key)
                .updateExpression(OBTAIN_LOCK_QUERY)
                .conditionExpression(OBTAIN_LOCK_CONDITION)
                .expressionAttributeValues(attributeUpdates)
                .returnValues(ReturnValue.UPDATED_NEW)
                .build();

        try {
            // There are three possible situations:
            // 1. The lock document does not exist yet - it is inserted - we have the lock
            // 2. The lock document exists and lockUtil <= now - it is updated - we have the
            // lock
            // 3. The lock document exists and lockUtil > now -
            // ConditionalCheckFailedException is thrown
            dynamoDbClient.updateItem(request);
            return Optional.of(new DynamoDBLock(dynamoDbClient, tableName, lockConfiguration));
        } catch (ConditionalCheckFailedException e) {
            // Condition failed. This means there was a lock with lockUntil > now.
            return Optional.empty();
        }
    
<no_super_class>