class ElasticsearchLockProvider implements LockProvider {
    static final String SCHEDLOCK_DEFAULT_INDEX = "shedlock";
    static final String LOCK_UNTIL = "lockUntil";
    static final String LOCKED_AT = "lockedAt";
    static final String LOCKED_BY = "lockedBy";
    static final String NAME = "name";

    private static final String UPDATE_SCRIPT = "if (ctx._source." + LOCK_UNTIL + " <= " + "params." + LOCKED_AT
            + ") { " + "ctx._source." + LOCKED_BY + " = params." + LOCKED_BY + "; " + "ctx._source." + LOCKED_AT
            + " = params." + LOCKED_AT + "; " + "ctx._source." + LOCK_UNTIL + " =  params." + LOCK_UNTIL + "; "
            + "} else { " + "ctx.op = 'none' " + "}";

    private final ElasticsearchClient client;
    private final String hostname;
    private final String index;

    private ElasticsearchLockProvider(@NonNull ElasticsearchClient client, @NonNull String index) {
        this.client = client;
        this.hostname = getHostname();
        this.index = index;
    }

    public ElasticsearchLockProvider(@NonNull ElasticsearchClient client) {
        this(client, SCHEDLOCK_DEFAULT_INDEX);
    }

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

    private Map<String, JsonData> lockObject(String name, Instant lockUntil, Instant lockedAt) {
        return Map.of(
                NAME,
                JsonData.of(name),
                LOCKED_BY,
                JsonData.of(hostname),
                LOCKED_AT,
                JsonData.of(lockedAt.toEpochMilli()),
                LOCK_UNTIL,
                JsonData.of(lockUntil.toEpochMilli()));
    }

    private final class ElasticsearchSimpleLock extends AbstractSimpleLock {

        private ElasticsearchSimpleLock(LockConfiguration lockConfiguration) {
            super(lockConfiguration);
        }

        @Override
        public void doUnlock() {
            // Set lockUtil to now or lockAtLeastUntil whichever is later
            try {
                Map<String, JsonData> lockObject = Collections.singletonMap(
                        "unlockTime",
                        JsonData.of(lockConfiguration.getUnlockTime().toEpochMilli()));

                UpdateRequest<Lock, Lock> updateRequest = UpdateRequest.of(ur -> ur.index(index)
                        .id(lockConfiguration.getName())
                        .refresh(Refresh.True)
                        .script(sc -> sc.inline(in -> in.lang("painless")
                                .source("ctx._source.lockUntil = params.unlockTime")
                                .params(lockObject))));
                client.update(updateRequest, Lock.class);
            } catch (IOException | ElasticsearchException e) {
                throw new LockException("Unexpected exception occurred", e);
            }
        }
    }

    private static final class Lock {

        private final String name;
        private final String lockedBy;
        private final long lockedAt;
        private final long lockUntil;

        public Lock(String name, String lockedBy, Instant lockedAt, Instant lockUntil) {
            this.name = name;
            this.lockedBy = lockedBy;
            this.lockedAt = lockedAt.toEpochMilli();
            this.lockUntil = lockUntil.toEpochMilli();
        }

        public String getName() {
            return name;
        }

        public String getLockedBy() {
            return lockedBy;
        }

        public long getLockedAt() {
            return lockedAt;
        }

        public long getLockUntil() {
            return lockUntil;
        }
    }
}

        try {
            Instant now = now();
            Instant lockAtMostUntil = lockConfiguration.getLockAtMostUntil();
            Map<String, JsonData> lockObject = lockObject(lockConfiguration.getName(), lockAtMostUntil, now);

            // The object exist only to have some type we can work with
            Lock pojo = new Lock(lockConfiguration.getName(), hostname, now, lockAtMostUntil);

            UpdateRequest<Lock, Lock> updateRequest = UpdateRequest.of(ur -> ur.index(index)
                    .id(lockConfiguration.getName())
                    .refresh(Refresh.True)
                    .script(sc -> sc.inline(
                            in -> in.lang("painless").source(UPDATE_SCRIPT).params(lockObject)))
                    .upsert(pojo));

            UpdateResponse<Lock> res = client.update(updateRequest, Lock.class);
            if (res.result() != Result.NoOp) {
                return Optional.of(new ElasticsearchSimpleLock(lockConfiguration));
            } else { // nothing happened
                return Optional.empty();
            }
        } catch (IOException | ElasticsearchException e) {
            if ((e instanceof ElasticsearchException && ((ElasticsearchException) e).status() == 409)
                    || (e instanceof ResponseException
                            && ((ResponseException) e)
                                            .getResponse()
                                            .getStatusLine()
                                            .getStatusCode()
                                    == 409)) {
                return Optional.empty();
            } else {
                throw new LockException("Unexpected exception occurred", e);
            }
        }
    
<no_super_class>