class CassandraStorageAccessor extends AbstractStorageAccessor {
    private final String hostname;
    private final CqlIdentifier table;
    private final CqlIdentifier keyspace;
    private final String lockName;
    private final String lockUntil;
    private final String lockedAt;
    private final String lockedBy;
    private final CqlSession cqlSession;
    private final ConsistencyLevel consistencyLevel;
    private final ConsistencyLevel serialConsistencyLevel;

    CassandraStorageAccessor(@NonNull Configuration configuration) {
        requireNonNull(configuration, "configuration can not be null");
        this.hostname = Utils.getHostname();
        this.table = configuration.getTable();
        this.keyspace = configuration.getKeyspace();
        this.lockName = configuration.getColumnNames().getLockName();
        this.lockUntil = configuration.getColumnNames().getLockUntil();
        this.lockedAt = configuration.getColumnNames().getLockedAt();
        this.lockedBy = configuration.getColumnNames().getLockedBy();
        this.cqlSession = configuration.getCqlSession();
        this.consistencyLevel = configuration.getConsistencyLevel();
        this.serialConsistencyLevel = configuration.getSerialConsistencyLevel();
    }

    @Override
    public boolean insertRecord(@NonNull LockConfiguration lockConfiguration) {
        if (find(lockConfiguration.getName()).isPresent()) {
            return false;
        }

        try {
            return insert(lockConfiguration.getName(), lockConfiguration.getLockAtMostUntil());
        } catch (QueryExecutionException e) {
            logger.warn("Error on insert", e);
            return false;
        }
    }

    @Override
    public boolean updateRecord(@NonNull LockConfiguration lockConfiguration) {
        Optional<Lock> lock = find(lockConfiguration.getName());
        if (lock.isEmpty() || lock.get().lockUntil().isAfter(ClockProvider.now())) {
            return false;
        }

        try {
            return update(lockConfiguration.getName(), lockConfiguration.getLockAtMostUntil());
        } catch (QueryExecutionException e) {
            logger.warn("Error on update", e);
            return false;
        }
    }

    @Override
    public void unlock(@NonNull LockConfiguration lockConfiguration) {
        updateUntil(lockConfiguration.getName(), lockConfiguration.getUnlockTime());
    }

    @Override
    public boolean extend(@NonNull LockConfiguration lockConfiguration) {
        Optional<Lock> lock = find(lockConfiguration.getName());
        if (lock.isEmpty()
                || lock.get().lockUntil().isBefore(ClockProvider.now())
                || !lock.get().lockedBy().equals(hostname)) {
            logger.trace("extend false");
            return false;
        }

        return updateUntil(lockConfiguration.getName(), lockConfiguration.getLockAtMostUntil());
    }

    /**
     * Find existing row by primary key lock.name
     *
     * @param name
     *            lock name
     * @return optional lock row or empty
     */
    Optional<Lock> find(String name) {<FILL_FUNCTION_BODY>}

    /**
     * Insert new lock row
     *
     * @param name
     *            lock name
     * @param until
     *            new until instant value
     */
    private boolean insert(String name, Instant until) {
        return execute(QueryBuilder.insertInto(keyspace, table)
                .value(lockName, literal(name))
                .value(lockUntil, literal(until))
                .value(lockedAt, literal(ClockProvider.now()))
                .value(lockedBy, literal(hostname))
                .ifNotExists()
                .build());
    }

    /**
     * Update existing lock row
     *
     * @param name
     *            lock name
     * @param until
     *            new until instant value
     */
    private boolean update(String name, Instant until) {
        return execute(QueryBuilder.update(keyspace, table)
                .setColumn(lockUntil, literal(until))
                .setColumn(lockedAt, literal(ClockProvider.now()))
                .setColumn(lockedBy, literal(hostname))
                .whereColumn(lockName)
                .isEqualTo(literal(name))
                .ifColumn(lockUntil)
                .isLessThan(literal(ClockProvider.now()))
                .build());
    }

    /**
     * Updates lock.until field where lockConfiguration.name
     *
     * @param name
     *            lock name
     * @param until
     *            new until instant value
     */
    private boolean updateUntil(String name, Instant until) {
        return execute(QueryBuilder.update(keyspace, table)
                .setColumn(lockUntil, literal(until))
                .whereColumn(lockName)
                .isEqualTo(literal(name))
                .ifColumn(lockUntil)
                .isGreaterThanOrEqualTo(literal(ClockProvider.now()))
                .ifColumn(lockedBy)
                .isEqualTo(literal(hostname))
                .build());
    }

    private boolean execute(SimpleStatement statement) {
        return cqlSession
                .execute(statement
                        .setConsistencyLevel(consistencyLevel)
                        .setSerialConsistencyLevel(serialConsistencyLevel))
                .wasApplied();
    }
}

        SimpleStatement selectStatement = QueryBuilder.selectFrom(keyspace, table)
                .column(lockUntil)
                .column(lockedAt)
                .column(lockedBy)
                .whereColumn(lockName)
                .isEqualTo(literal(name))
                .build()
                .setConsistencyLevel(consistencyLevel)
                .setSerialConsistencyLevel(serialConsistencyLevel);

        ResultSet resultSet = cqlSession.execute(selectStatement);
        Row row = resultSet.one();
        if (row != null) {
            return Optional.of(new Lock(row.getInstant(lockUntil), row.getInstant(lockedAt), row.getString(lockedBy)));
        } else {
            return Optional.empty();
        }
    
<methods>public non-sealed void <init>() <variables>protected final Logger logger