class ArangoLockProvider implements LockProvider {

    static final String LOCK_UNTIL = "lockUntil";
    static final String LOCKED_AT = "lockedAt";
    static final String LOCKED_BY = "lockedBy";
    static final String COLLECTION_NAME = "shedLock";

    private final ArangoCollection arangoCollection;

    /**
     * Instantiates a new Arango lock provider.
     *
     * @param arangoDatabase
     *            the arango database
     */
    public ArangoLockProvider(@NonNull ArangoDatabase arangoDatabase) {
        this(arangoDatabase.collection(COLLECTION_NAME));
    }

    /**
     * Instantiates a new Arango lock provider.
     *
     * @param arangoCollection
     *            the arango collection
     */
    public ArangoLockProvider(@NonNull ArangoCollection arangoCollection) {
        this.arangoCollection = arangoCollection;
    }

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

    private BaseDocument insertNewLock(String transactionId, String documentKey, Instant lockAtMostUntil)
            throws ArangoDBException {
        BaseDocument newDocument = new BaseDocument();
        newDocument.setKey(documentKey);
        setDocumentAttributes(newDocument, lockAtMostUntil);
        DocumentCreateEntity<BaseDocument> document = arangoCollection.insertDocument(
                newDocument,
                new DocumentCreateOptions().streamTransactionId(transactionId).returnNew(true));
        return document.getNew();
    }

    private void updateLockAtMostUntil(String transactionId, BaseDocument existingDocument, Instant lockAtMostUntil) {
        setDocumentAttributes(existingDocument, lockAtMostUntil);
        arangoCollection.updateDocument(
                existingDocument.getKey(),
                existingDocument,
                new DocumentUpdateOptions().streamTransactionId(transactionId));
    }

    private void setDocumentAttributes(BaseDocument baseDocument, Instant lockAtMostUntil) {
        baseDocument.addAttribute(LOCK_UNTIL, Utils.toIsoString(lockAtMostUntil));
        baseDocument.addAttribute(LOCKED_AT, Utils.toIsoString(ClockProvider.now()));
        baseDocument.addAttribute(LOCKED_BY, getHostname());
    }

    private static final class ArangoLock extends AbstractSimpleLock {

        private final ArangoCollection arangoCollection;
        private final BaseDocument document;

        public ArangoLock(
                final ArangoCollection arangoCollection,
                final BaseDocument document,
                final LockConfiguration lockConfiguration) {

            super(lockConfiguration);
            this.arangoCollection = arangoCollection;
            this.document = document;
        }

        @Override
        protected void doUnlock() {
            try {
                document.addAttribute(LOCK_UNTIL, Utils.toIsoString(lockConfiguration.getUnlockTime()));
                arangoCollection.updateDocument(lockConfiguration.getName(), document);
            } catch (ArangoDBException e) {
                throw new LockException("Unexpected error occured", e);
            }
        }
    }
}

        String transactionId = null;

        try {
            /*
             * Transaction is necessary because repsert (insert with overwrite=true in
             * arangodb) is not possible with condition check (see case 2 description below)
             */
            StreamTransactionEntity streamTransactionEntity = arangoCollection
                    .db()
                    .beginStreamTransaction(
                            new StreamTransactionOptions().exclusiveCollections(arangoCollection.name()));

            transactionId = streamTransactionEntity.getId();

            /*
             * There are three possible situations: 1. The lock document does not exist yet
             * - insert document 2. The lock document exists and lockUtil <= now - update
             * document 3. The lock document exists and lockUtil > now - nothing to do
             */
            BaseDocument existingDocument = arangoCollection.getDocument(
                    lockConfiguration.getName(),
                    BaseDocument.class,
                    new DocumentReadOptions().streamTransactionId(transactionId));

            // 1. case
            if (existingDocument == null) {
                BaseDocument newDocument = insertNewLock(
                        transactionId, lockConfiguration.getName(), lockConfiguration.getLockAtMostUntil());
                return Optional.of(new ArangoLock(arangoCollection, newDocument, lockConfiguration));
            }

            // 2. case
            Instant lockUntil =
                    Instant.parse(existingDocument.getAttribute(LOCK_UNTIL).toString());
            if (lockUntil.compareTo(ClockProvider.now()) <= 0) {
                updateLockAtMostUntil(transactionId, existingDocument, lockConfiguration.getLockAtMostUntil());
                return Optional.of(new ArangoLock(arangoCollection, existingDocument, lockConfiguration));
            }

            // 3. case
            return Optional.empty();

        } catch (ArangoDBException e) {
            if (transactionId != null) {
                arangoCollection.db().abortStreamTransaction(transactionId);
            }
            throw new LockException("Unexpected error occured", e);
        } finally {
            if (transactionId != null) {
                arangoCollection.db().commitStreamTransaction(transactionId);
            }
        }
    
<no_super_class>