Copilot commented on code in PR #3695:
URL: https://github.com/apache/celeborn/pull/3695#discussion_r3263511187
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
}
Review Comment:
Re-creating the DB on any `RocksDBException` is dangerous because
`RocksDBProvider.initRockDB` interprets an open failure as corruption and
deletes the entire RocksDB directory (see `RocksDBProvider.java` lines 84–100)
before creating an empty one. If a put/get/delete throws for any reason
(transient I/O error, lock contention, etc.) and the subsequent reopen happens
to fail, all worker metadata will be silently wiped instead of preserved for
manual recovery. Auto-recovery should at minimum avoid the wipe-and-recreate
branch, or be restricted to specific RocksDB status codes (e.g., IOError) that
are known to be recoverable by reopen.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
Review Comment:
If `RocksDBProvider.initRockDB` throws `IOException`, the error is logged
but `dbGeneration` is not incremented and the old (now-closed) `db` reference
is retained. Every subsequent operation will read the same generation, fail
against the closed DB, and re-enter `recreateDBInstance` — turning each request
into another full reopen attempt under the write lock. This is both a tight
retry loop and a serialization point under failure. Consider incrementing the
generation regardless of success, or setting a permanent failure state that
fast-fails operations.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value) throws RocksDBException
{
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.put(key, value);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
Review Comment:
After `recreateDBInstance` successfully reopens the DB, the original
`RocksDBException` is still rethrown to the caller. So even though the PR's
stated goal is that "metadata operations can recover automatically and continue
functioning", the failing operation is not retried and surfaces as a failure to
callers, who must implement their own retry logic. Either document this
contract explicitly or transparently retry the operation once on the new
generation.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value) throws RocksDBException
{
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.put(key, value);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value, boolean sync) throws
RocksDBException {
- if (sync) {
- db.put(SYNC_WRITE_OPTIONS, key, value);
- } else {
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ if (sync) {
+ db.put(SYNC_WRITE_OPTIONS, key, value);
+ } else {
+ db.put(key, value);
+ }
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
}
}
@Override
protected byte[] getInternal(byte[] key) throws RocksDBException {
- return db.get(key);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ return db.get(key);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected void deleteInternal(byte[] key) throws RocksDBException {
- db.delete(key);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.delete(key);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected DBIterator newIterator(MetadataMetrics metrics) {
- return new RocksDBIterator(db.newIterator(), metrics);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ return new RocksDBIterator(db.newIterator(), metrics);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RuntimeException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
Review Comment:
`newIterator` returns a `RocksDBIterator` that holds a reference to the
underlying native iterator of the current `db`. After this method returns, the
read lock is released and the iterator is then used by callers without any lock
protection. If `recreateDBInstance` runs concurrently and closes the previous
`db`, in-flight iterators reference freed native handles, leading to JVM
crashes or undefined behavior. Recovery must either invalidate outstanding
iterators or be deferred while iterators are alive.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value) throws RocksDBException
{
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.put(key, value);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value, boolean sync) throws
RocksDBException {
- if (sync) {
- db.put(SYNC_WRITE_OPTIONS, key, value);
- } else {
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ if (sync) {
+ db.put(SYNC_WRITE_OPTIONS, key, value);
+ } else {
+ db.put(key, value);
+ }
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
}
}
@Override
protected byte[] getInternal(byte[] key) throws RocksDBException {
- return db.get(key);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ return db.get(key);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected void deleteInternal(byte[] key) throws RocksDBException {
- db.delete(key);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.delete(key);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected DBIterator newIterator(MetadataMetrics metrics) {
- return new RocksDBIterator(db.newIterator(), metrics);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ return new RocksDBIterator(db.newIterator(), metrics);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RuntimeException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
public void close() throws IOException {
+ rwLock.writeLock().lock();
try {
db.close();
} finally {
- // WriteOptions is a native handle; release it even if db.close() throws.
+ rwLock.writeLock().unlock();
SYNC_WRITE_OPTIONS.close();
}
}
Review Comment:
`close()` acquires the write lock and closes `db` but does not null it out
or set a "closed" flag. If a put/get/delete is called after `close()`, the
operation will throw, fall through into the `catch (RocksDBException e)`
branch, and `recreateDBInstance` will happily reopen the database via
`RocksDBProvider.initRockDB`, resurrecting an instance that was explicitly
closed and leaking native resources. Consider tracking a closed state and
short-circuiting `recreateDBInstance`/all operations after `close()`.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
}
Review Comment:
No tests are added for this new automatic recovery behavior, even though it
is non-trivial (concurrent recreate via generation counter, write/read lock
ordering, exception path). Given the data-loss risk in the recreate path and
the concurrency model, at minimum unit tests covering: (1) successful recreate
on simulated `RocksDBException`, (2) only one recreate when multiple threads
fail concurrently with the same generation, and (3) operations after `close()`
not resurrecting the DB, would significantly reduce risk.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value) throws RocksDBException
{
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.put(key, value);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
Review Comment:
The read lock is acquired inside the outer `try` block. If the `lock()` call
itself were to throw (e.g., via thread interruption with a future lock
variant), the `finally { rwLock.readLock().unlock(); }` would still run and
attempt to unlock a lock that was never acquired, throwing
`IllegalMonitorStateException` and masking the original cause. The conventional
pattern is to call `lock()` before the `try` block. This pattern is duplicated
in `putInternal` (both overloads), `getInternal`, `deleteInternal`, and
`newIterator`.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDB.java:
##########
@@ -30,49 +36,154 @@
* <p>Note: code copied from Apache Spark.
*/
public class RocksDB extends DB {
- private final org.rocksdb.RocksDB db;
+ private static final Logger logger = LoggerFactory.getLogger(RocksDB.class);
+
+ private volatile org.rocksdb.RocksDB db;
private final WriteOptions SYNC_WRITE_OPTIONS = new
WriteOptions().setSync(true);
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final AtomicLong dbGeneration = new AtomicLong(0);
+ private final File dbFile;
+ private final StoreVersion version;
- public RocksDB(org.rocksdb.RocksDB db, AbstractSource source, DBBackend
dbBackend) {
+ public RocksDB(
+ org.rocksdb.RocksDB db,
+ AbstractSource source,
+ DBBackend dbBackend,
+ File dbFile,
+ StoreVersion version) {
super(source, dbBackend);
this.db = db;
+ this.dbFile = dbFile;
+ this.version = version;
+ }
+
+ private void recreateDBInstance(long failedGeneration) {
+ rwLock.writeLock().lock();
+ try {
+ if (dbGeneration.get() != failedGeneration) {
+ logger.info(
+ "RocksDB instance already recovered by another thread (generation
{} -> {})",
+ failedGeneration,
+ dbGeneration.get());
+ return;
+ }
+
+ try {
+ if (db != null) {
+ db.close();
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close RocksDB instance", e);
+ }
+ db = RocksDBProvider.initRockDB(dbFile, version);
+ dbGeneration.incrementAndGet();
+ } catch (IOException e) {
+ logger.error(
+ "Failed to recreate RocksDB instance at {}. "
+ + "Database is unavailable, all subsequent operations will
fail.",
+ dbFile,
+ e);
+ } finally {
+ rwLock.writeLock().unlock();
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value) throws RocksDBException
{
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.put(key, value);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected void putInternal(byte[] key, byte[] value, boolean sync) throws
RocksDBException {
- if (sync) {
- db.put(SYNC_WRITE_OPTIONS, key, value);
- } else {
- db.put(key, value);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ if (sync) {
+ db.put(SYNC_WRITE_OPTIONS, key, value);
+ } else {
+ db.put(key, value);
+ }
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
}
}
@Override
protected byte[] getInternal(byte[] key) throws RocksDBException {
- return db.get(key);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ return db.get(key);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected void deleteInternal(byte[] key) throws RocksDBException {
- db.delete(key);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ db.delete(key);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RocksDBException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
@Override
protected DBIterator newIterator(MetadataMetrics metrics) {
- return new RocksDBIterator(db.newIterator(), metrics);
+ long generation = 0;
+ try {
+ rwLock.readLock().lock();
+ generation = dbGeneration.get();
+ try {
+ return new RocksDBIterator(db.newIterator(), metrics);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ } catch (RuntimeException e) {
+ recreateDBInstance(generation);
+ throw e;
+ }
}
Review Comment:
The five operation methods all duplicate the same
try/lock/generation/recreate scaffolding. This boilerplate is error-prone
(e.g., easy to miss `unlock` or place `lock()` in the wrong scope) and will
need to be repeated for every future DB method. Consider extracting a private
`<T> T withRecovery(Callable<T>)` / `runWithRecovery(Runnable)` helper that
encapsulates the read-lock, generation snapshot, exception capture, and
recreate-on-failure logic.
--
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]
