gnodet commented on code in PR #25993: URL: https://github.com/apache/camel/pull/25993#discussion_r3914272660
########## components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java: ########## @@ -0,0 +1,439 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.processor.keyvalue.jdbc; + +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +import javax.sql.DataSource; + +import org.apache.camel.api.management.ManagedAttribute; +import org.apache.camel.api.management.ManagedOperation; +import org.apache.camel.api.management.ManagedResource; +import org.apache.camel.spi.Configurer; +import org.apache.camel.spi.KeyValueRepository; +import org.apache.camel.spi.Metadata; +import org.apache.camel.support.KeyValueRepositoryHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.dao.EmptyResultDataAccessException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.TransactionTemplate; + +/** + * A JDBC-based implementation of {@link KeyValueRepository} that stores entries in a database table. + * <p/> + * Values are serialized using Java object serialization and stored as BLOB. Each entry may optionally have a + * time-to-live (TTL); expired entries are lazily evicted on access and during {@link #keys()} scans. + * <p/> + * The table is created automatically on startup if it does not already exist (controlled by + * {@link #setCreateTableIfNotExists(boolean)}). + * + * @since 4.23 + */ +@Metadata(label = "bean", + description = "A JDBC-based KeyValueRepository that stores entries in a database table.", + annotations = { "interfaceName=org.apache.camel.spi.KeyValueRepository" }) +@Configurer(metadataOnly = true) +@ManagedResource(description = "JDBC based key-value repository") +public class JdbcKeyValueRepository extends ServiceSupport implements KeyValueRepository { + + protected static final String DEFAULT_TABLENAME = "CAMEL_KEYVALUE"; + protected static final String DEFAULT_TABLE_EXISTS_STRING = "SELECT 1 FROM CAMEL_KEYVALUE WHERE 1 = 0"; + protected static final String DEFAULT_CREATE_STRING + = "CREATE TABLE CAMEL_KEYVALUE (ITEM_KEY VARCHAR(512) NOT NULL, ITEM_VALUE BLOB NOT NULL, " + + "EXPIRES_AT BIGINT NOT NULL DEFAULT 0, PRIMARY KEY (ITEM_KEY))"; + protected static final String DEFAULT_SELECT_STRING + = "SELECT ITEM_VALUE, EXPIRES_AT FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ?"; + protected static final String DEFAULT_INSERT_STRING + = "INSERT INTO CAMEL_KEYVALUE (ITEM_KEY, ITEM_VALUE, EXPIRES_AT) VALUES (?, ?, ?)"; + protected static final String DEFAULT_DELETE_STRING = "DELETE FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ?"; + protected static final String DEFAULT_CLEAR_STRING = "DELETE FROM CAMEL_KEYVALUE"; + protected static final String DEFAULT_SELECT_KEYS_STRING + = "SELECT ITEM_KEY FROM CAMEL_KEYVALUE WHERE EXPIRES_AT = 0 OR EXPIRES_AT > ?"; + protected static final String DEFAULT_DELETE_EXPIRED_STRING + = "DELETE FROM CAMEL_KEYVALUE WHERE EXPIRES_AT > 0 AND EXPIRES_AT <= ?"; + + private static final Logger LOG = LoggerFactory.getLogger(JdbcKeyValueRepository.class); + + @Metadata(description = "The Spring JdbcTemplate to use for connecting to the database", required = true) + private JdbcTemplate jdbcTemplate; + @Metadata(description = "The Spring TransactionTemplate to use for connecting to the database", required = true) + private TransactionTemplate transactionTemplate; + private DataSource dataSource; + + @Metadata(description = "The name of the table to use in the database", defaultValue = "CAMEL_KEYVALUE") + private String tableName; + @Metadata(description = "Whether to create the table in the database if none exists on startup", defaultValue = "true") + private boolean createTableIfNotExists = true; + + @Metadata(label = "advanced", description = "SQL query to use for checking if table exists") + private String tableExistsString = DEFAULT_TABLE_EXISTS_STRING; + @Metadata(label = "advanced", description = "SQL query to use for creating table") + private String createString = DEFAULT_CREATE_STRING; + @Metadata(label = "advanced", description = "SQL query to use for selecting a value by key") + private String selectString = DEFAULT_SELECT_STRING; + @Metadata(label = "advanced", description = "SQL query to use for inserting a new entry") + private String insertString = DEFAULT_INSERT_STRING; + @Metadata(label = "advanced", description = "SQL query to use for deleting an entry by key") + private String deleteString = DEFAULT_DELETE_STRING; + @Metadata(label = "advanced", description = "SQL query to delete all entries from the table") + private String clearString = DEFAULT_CLEAR_STRING; + @Metadata(label = "advanced", description = "SQL query to use for selecting all non-expired keys") + private String selectKeysString = DEFAULT_SELECT_KEYS_STRING; + @Metadata(label = "advanced", description = "SQL query to use for deleting expired entries") + private String deleteExpiredString = DEFAULT_DELETE_EXPIRED_STRING; + + /** + * Creates a new JDBC key-value repository. A {@link DataSource} or {@link JdbcTemplate} must be set before + * initialization. + */ + public JdbcKeyValueRepository() { + } + + /** + * Creates a new JDBC key-value repository using the given data source. A {@link JdbcTemplate} and + * {@link TransactionTemplate} will be created automatically during initialization. + * + * @param dataSource the data source to use + */ + public JdbcKeyValueRepository(DataSource dataSource) { + this.dataSource = dataSource; + } + + /** + * Creates a new JDBC key-value repository using the given JDBC template and transaction template. + * + * @param jdbcTemplate the JDBC template for database access + * @param transactionTemplate the transaction template for transactional operations + */ + public JdbcKeyValueRepository(JdbcTemplate jdbcTemplate, TransactionTemplate transactionTemplate) { + this.jdbcTemplate = jdbcTemplate; + this.transactionTemplate = transactionTemplate; + } + + /** + * Creates a {@link TransactionTemplate} from the given data source with {@code PROPAGATION_REQUIRED}. + * + * @param dataSource the data source to create the transaction template from + * @return a configured transaction template + */ + protected static TransactionTemplate createTransactionTemplate(DataSource dataSource) { + TransactionTemplate transactionTemplate = new TransactionTemplate(); + transactionTemplate.setTransactionManager(new DataSourceTransactionManager(dataSource)); + transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED); + return transactionTemplate; + } + + @Override + protected void doInit() throws Exception { + super.doInit(); + + if (dataSource != null && jdbcTemplate == null) { + jdbcTemplate = new JdbcTemplate(dataSource); + jdbcTemplate.afterPropertiesSet(); + } + if (dataSource != null && transactionTemplate == null) { + transactionTemplate = createTransactionTemplate(dataSource); + } + + if (tableName != null) { + // update query strings from default table name to the custom table name + tableExistsString = DEFAULT_TABLE_EXISTS_STRING.replace(DEFAULT_TABLENAME, tableName); + createString = DEFAULT_CREATE_STRING.replace(DEFAULT_TABLENAME, tableName); + selectString = DEFAULT_SELECT_STRING.replace(DEFAULT_TABLENAME, tableName); + insertString = DEFAULT_INSERT_STRING.replace(DEFAULT_TABLENAME, tableName); + deleteString = DEFAULT_DELETE_STRING.replace(DEFAULT_TABLENAME, tableName); + clearString = DEFAULT_CLEAR_STRING.replace(DEFAULT_TABLENAME, tableName); + selectKeysString = DEFAULT_SELECT_KEYS_STRING.replace(DEFAULT_TABLENAME, tableName); + deleteExpiredString = DEFAULT_DELETE_EXPIRED_STRING.replace(DEFAULT_TABLENAME, tableName); + } + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + + boolean tableExists = transactionTemplate.execute(status -> { + try { + // we will receive an exception if the table doesn't exist or we cannot access it + jdbcTemplate.execute(getTableExistsString()); + LOG.debug("Expected table for JdbcKeyValueRepository exists"); + return true; + } catch (DataAccessException e) { + LOG.debug("Expected table for JdbcKeyValueRepository does not exist"); + return false; + } + }); + + if (!tableExists && createTableIfNotExists) { + transactionTemplate.executeWithoutResult(status -> { + try { + LOG.debug("Creating table for JdbcKeyValueRepository because it doesn't exist..."); + jdbcTemplate.execute(getCreateString()); + LOG.info("Table created with query '{}'", getCreateString()); + } catch (DataAccessException dae) { + LOG.error( + "Can't create table for JdbcKeyValueRepository with query '{}' because of: {}. " + + "This may be a permissions problem. Please create this table and try again.", + getCreateString(), dae.getMessage()); + throw dae; + } + }); + } + } + + @Override + protected void doStop() throws Exception { + // noop + } + + @Override + @ManagedOperation(description = "Get value by key") + public Object get(String key) { + return transactionTemplate.execute(status -> doGet(key)); + } + + @Override + @ManagedOperation(description = "Put a key-value pair with optional TTL") + public Object put(String key, Object value, Duration ttl) { + return transactionTemplate.execute(status -> { + Object oldValue = doGet(key); + // delete any existing row (whether expired or not) + jdbcTemplate.update(getDeleteString(), key); + // insert the new row + long expiresAt = toExpiresAt(ttl); + jdbcTemplate.update(getInsertString(), key, KeyValueRepositoryHelper.serialize(value), expiresAt); + return oldValue; + }); + } + + @Override + @ManagedOperation(description = "Delete a key") + public Object delete(String key) { + return transactionTemplate.execute(status -> { + Object oldValue = doGet(key); + jdbcTemplate.update(getDeleteString(), key); + return oldValue; + }); + } + + @Override + @ManagedOperation(description = "Check if key exists") + public boolean contains(String key) { + Boolean result = transactionTemplate.execute(status -> doGet(key) != null); + return result != null && result; + } + + @Override + public Set<String> keys() { + return transactionTemplate.execute(status -> { + // first delete expired entries + long now = System.currentTimeMillis(); + jdbcTemplate.update(getDeleteExpiredString(), now); + // then select all non-expired keys + List<String> keyList = jdbcTemplate.queryForList(getSelectKeysString(), String.class, now); + return Collections.unmodifiableSet(new LinkedHashSet<>(keyList)); + }); + } + + @Override + @ManagedOperation(description = "Clear all entries") + public void clear() { + transactionTemplate.executeWithoutResult(status -> jdbcTemplate.update(getClearString())); + } + + @Override + public Object putIfAbsent(String key, Object value, Duration ttl) { + return transactionTemplate.execute(status -> { + // check if a non-expired entry already exists + Object existing = doGet(key); + if (existing != null) { + return existing; + } + // attempt to insert + long expiresAt = toExpiresAt(ttl); + try { + jdbcTemplate.update(getInsertString(), key, KeyValueRepositoryHelper.serialize(value), expiresAt); + return null; + } catch (DuplicateKeyException e) { + // concurrent insert race -- another thread/node won + LOG.debug("Concurrent insert race for key '{}' -- another thread won, treating as existing", key); + status.setRollbackOnly(); + // re-read the value that the other thread inserted + Object concurrentValue = doGet(key); + return concurrentValue; + } + }); + } + + @Override + @ManagedAttribute(description = "The number of entries in the repository") + public int size() { + return keys().size(); + } + + /** + * Internal get that reads the value for the given key within the current transaction. If the entry has expired, it + * is deleted and {@code null} is returned. + * + * @param key the key to look up + * @return the deserialized value, or {@code null} if not found or expired + */ + private Object doGet(String key) { + try { + return jdbcTemplate.queryForObject(getSelectString(), (rs, rowNum) -> { + byte[] bytes = rs.getBytes(1); + long expiresAt = rs.getLong(2); + if (expiresAt > 0 && System.currentTimeMillis() >= expiresAt) { + // entry has expired -- delete it + jdbcTemplate.update(getDeleteString(), key); Review Comment: _Claude Code on behalf of gnodet_ Good catch. Refactored `doGet()` to use a sentinel flag: the RowMapper now sets an `expired` flag instead of executing the DELETE while the ResultSet is open. The expired-row cleanup runs after `queryForObject` returns, avoiding the potential issue with JDBC drivers that dislike concurrent statements on the same connection. ########## components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java: ########## @@ -0,0 +1,369 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.processor.keyvalue.cassandra; + +import java.nio.ByteBuffer; +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.Set; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.cql.PreparedStatement; +import com.datastax.oss.driver.api.core.cql.ResultSet; +import com.datastax.oss.driver.api.core.cql.Row; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import com.datastax.oss.driver.api.querybuilder.delete.Delete; +import com.datastax.oss.driver.api.querybuilder.select.Select; +import com.datastax.oss.driver.api.querybuilder.truncate.Truncate; +import org.apache.camel.api.management.ManagedAttribute; +import org.apache.camel.api.management.ManagedOperation; +import org.apache.camel.api.management.ManagedResource; +import org.apache.camel.spi.Configurer; +import org.apache.camel.spi.KeyValueRepository; +import org.apache.camel.spi.Metadata; +import org.apache.camel.support.KeyValueRepositoryHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.apache.camel.util.ObjectHelper; +import org.apache.camel.utils.cassandra.CassandraSessionHolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker; +import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.insertInto; +import static org.apache.camel.utils.cassandra.CassandraUtils.applyConsistencyLevel; +import static org.apache.camel.utils.cassandra.CassandraUtils.generateDelete; +import static org.apache.camel.utils.cassandra.CassandraUtils.generateSelect; +import static org.apache.camel.utils.cassandra.CassandraUtils.generateTruncate; + +/** + * A Cassandra-based implementation of {@link KeyValueRepository} that stores key-value entries in a Cassandra table. + * <p/> + * Values are serialized to bytes using Java {@link ObjectOutputStream} and stored in a {@code BLOB} column. Keys are + * stored as {@code TEXT}. Time-to-live is handled natively by Cassandra's {@code USING TTL} clause on {@code INSERT} + * statements, so expired entries are removed automatically by Cassandra without any client-side eviction logic. + * <p/> + * The {@link #putIfAbsent(String, Object, long)} method is implemented atomically using Cassandra's lightweight + * transactions ({@code INSERT ... IF NOT EXISTS}). + * <p/> + * Advice: use LeveledCompaction for the backing table and tune read/write consistency levels for your use case. + * + * @since 4.23 + */ +@Metadata(label = "bean", + description = "A Cassandra-based KeyValueRepository that uses a Cassandra table to store key-value entries." + + " Advice: use LeveledCompaction for this table and tune read/write consistency levels.", + annotations = { "interfaceName=org.apache.camel.spi.KeyValueRepository" }) +@Configurer(metadataOnly = true) +@ManagedResource(description = "Cassandra based key-value repository") +public class CassandraKeyValueRepository extends ServiceSupport implements KeyValueRepository { + + private static final Logger LOGGER = LoggerFactory.getLogger(CassandraKeyValueRepository.class); + + private static final String KEY_COLUMN = "key"; + private static final String VALUE_COLUMN = "value"; + + @Metadata(description = "Cassandra session", required = true) + private CassandraSessionHolder session; + @Metadata(description = "The table name for storing the data", defaultValue = "camel_keyvalue") + private String table = "camel_keyvalue"; + @Metadata(description = "Write consistency level", + enums = "ANY,ONE,TWO,THREE,QUORUM,ALL,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,SERIAL,LOCAL_SERIAL") + private ConsistencyLevel writeConsistencyLevel; + @Metadata(description = "Read consistency level", + enums = "ANY,ONE,TWO,THREE,QUORUM,ALL,LOCAL_ONE,LOCAL_QUORUM,EACH_QUORUM,SERIAL,LOCAL_SERIAL") + private ConsistencyLevel readConsistencyLevel; + + private PreparedStatement insertStatement; + private PreparedStatement insertWithTtlStatement; + private PreparedStatement selectStatement; + private PreparedStatement deleteStatement; + private PreparedStatement selectAllKeysStatement; + private PreparedStatement truncateStatement; + private PreparedStatement insertIfNotExistsStatement; + private PreparedStatement insertIfNotExistsWithTtlStatement; + + public CassandraKeyValueRepository() { + } + + public CassandraKeyValueRepository(CqlSession session) { + this.session = new CassandraSessionHolder(session); + } + + // ------------------------------------------------------------------------- + // Helper methods + + /** + * Checks whether a lightweight transaction was applied. + * + * @param resultSet the result set from a conditional statement + * @return {@code true} if the statement was applied or the result is empty + */ + protected final boolean isApplied(ResultSet resultSet) { + Row row = resultSet.one(); + return row == null || row.getBoolean("[applied]"); + } + + // ------------------------------------------------------------------------- + // Lifecycle methods + + @Override + protected void doStart() throws Exception { + ObjectHelper.notNull(session, "session", this); + session.start(); + initInsertStatement(); + initInsertWithTtlStatement(); + initSelectStatement(); + initDeleteStatement(); + initSelectAllKeysStatement(); + initClearStatement(); + initInsertIfNotExistsStatement(); + initInsertIfNotExistsWithTtlStatement(); + } + + @Override + protected void doStop() throws Exception { + if (session != null) { + session.stop(); + } + } + + // ------------------------------------------------------------------------- + // Prepared statement initialization + + protected void initInsertStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert {}", statement); + insertStatement = getSession().prepare(statement); + } + + protected void initInsertWithTtlStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .usingTtl(bindMarker()) + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert with TTL {}", statement); + insertWithTtlStatement = getSession().prepare(statement); + } + + protected void initSelectStatement() { + Select select = generateSelect(table, new String[] { VALUE_COLUMN }, new String[] { KEY_COLUMN }); + SimpleStatement statement = applyConsistencyLevel(select.build(), readConsistencyLevel); + LOGGER.debug("Generated Select {}", statement); + selectStatement = getSession().prepare(statement); + } + + protected void initDeleteStatement() { + Delete delete = generateDelete(table, new String[] { KEY_COLUMN }, true); + SimpleStatement statement = applyConsistencyLevel(delete.build(), writeConsistencyLevel); + LOGGER.debug("Generated Delete {}", statement); + deleteStatement = getSession().prepare(statement); + } + + protected void initSelectAllKeysStatement() { + Select select = generateSelect(table, new String[] { KEY_COLUMN }, null); + SimpleStatement statement = applyConsistencyLevel(select.build(), readConsistencyLevel); + LOGGER.debug("Generated Select all keys {}", statement); + selectAllKeysStatement = getSession().prepare(statement); + } + + protected void initClearStatement() { + Truncate truncate = generateTruncate(table); + SimpleStatement statement = applyConsistencyLevel(truncate.build(), writeConsistencyLevel); + LOGGER.debug("Generated truncate for clear operation {}", statement); + truncateStatement = getSession().prepare(statement); + } + + protected void initInsertIfNotExistsStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .ifNotExists() + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert if not exists {}", statement); + insertIfNotExistsStatement = getSession().prepare(statement); + } + + protected void initInsertIfNotExistsWithTtlStatement() { + SimpleStatement statement = applyConsistencyLevel( + insertInto(table) + .value(KEY_COLUMN, bindMarker()) + .value(VALUE_COLUMN, bindMarker()) + .ifNotExists() + .usingTtl(bindMarker()) + .build(), + writeConsistencyLevel); + LOGGER.debug("Generated Insert if not exists with TTL {}", statement); + insertIfNotExistsWithTtlStatement = getSession().prepare(statement); + } + + // ------------------------------------------------------------------------- + // KeyValueRepository operations + + @Override + @ManagedOperation(description = "Get value by key") + public Object get(String key) { + LOGGER.debug("Getting key {}", key); + ResultSet rs = getSession().execute(selectStatement.bind(key)); + Row row = rs.one(); + if (row == null) { + return null; + } + ByteBuffer buffer = row.getByteBuffer(VALUE_COLUMN); + return buffer != null ? KeyValueRepositoryHelper.deserialize(buffer) : null; + } + + @Override + @ManagedOperation(description = "Put a key-value pair with optional TTL") + public Object put(String key, Object value, Duration ttl) { + LOGGER.debug("Putting key {} with TTL {}", key, ttl); + // Read the previous value before upserting + Object oldValue = get(key); + ByteBuffer serializedValue = KeyValueRepositoryHelper.serializeToByteBuffer(value); + int ttlSeconds = toTtlSeconds(ttl); + if (ttlSeconds > 0) { + getSession().execute(insertWithTtlStatement.bind(key, serializedValue, ttlSeconds)); + } else { + getSession().execute(insertStatement.bind(key, serializedValue)); + } + return oldValue; + } + + @Override + @ManagedOperation(description = "Delete a key") + public Object delete(String key) { + LOGGER.debug("Deleting key {}", key); + // Read the previous value before deleting + Object oldValue = get(key); + getSession().execute(deleteStatement.bind(key)); + return oldValue; + } + + @Override + @ManagedOperation(description = "Check if key exists") + public boolean contains(String key) { + LOGGER.debug("Checking key {}", key); + ResultSet rs = getSession().execute(selectStatement.bind(key)); + return rs.one() != null; + } + + @Override + public Set<String> keys() { + LOGGER.debug("Getting all keys from table {}", table); + ResultSet rs = getSession().execute(selectAllKeysStatement.bind()); + Set<String> result = new LinkedHashSet<>(); + for (Row row : rs) { + result.add(row.getString(KEY_COLUMN)); + } + return Collections.unmodifiableSet(result); + } + + @Override + @ManagedOperation(description = "Clear all entries") + public void clear() { + LOGGER.debug("Clear table {}", table); + getSession().execute(truncateStatement.bind()); + } + + /** + * Atomically stores the value under the given key only if no mapping already exists, using Cassandra's lightweight + * transaction ({@code INSERT ... IF NOT EXISTS}). + * + * @param key the key + * @param value the value to store + * @param ttl the time-to-live; {@code null}, zero, or negative means no expiration + * @return the existing value if the key was already present, or {@code null} if the put succeeded + */ + @Override + public Object putIfAbsent(String key, Object value, Duration ttl) { Review Comment: _Claude Code on behalf of gnodet_ Agreed — this is a natural follow-up. Cassandra's `UPDATE ... IF value = ?` and `DELETE ... IF value = ?` would give true CAS semantics for `replace()` and `delete(key, expected)`. Keeping it out of scope for this PR since the SPI default works correctly (non-atomically), and the manual documents it as "No" for atomic CAS on this backend. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
