Copilot commented on code in PR #3695:
URL: https://github.com/apache/celeborn/pull/3695#discussion_r3278304989
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDBProvider.java:
##########
@@ -48,26 +48,49 @@ public class RocksDBProvider {
private static final Logger logger =
LoggerFactory.getLogger(RocksDBProvider.class);
+ private static Options createDBOptions() {
+ BloomFilter fullFilter = new BloomFilter(10.0D /*
BloomFilter.DEFAULT_BITS_PER_KEY */, false);
+ BlockBasedTableConfig tableFormatConfig =
+ new BlockBasedTableConfig()
+ .setFilterPolicy(fullFilter)
+ .setEnableIndexCompression(false)
+ .setIndexBlockRestartInterval(8)
+ .setFormatVersion(5);
+
+ Options dbOptions = new Options();
+ RocksDBLogger rocksDBLogger = new RocksDBLogger(dbOptions);
+
+ dbOptions.setCreateIfMissing(false);
+ dbOptions.setBottommostCompressionType(CompressionType.ZSTD_COMPRESSION);
+ dbOptions.setCompressionType(CompressionType.LZ4_COMPRESSION);
+ dbOptions.setTableFormatConfig(tableFormatConfig);
+ dbOptions.setLogger(rocksDBLogger);
+
+ return dbOptions;
+ }
+
+ /**
+ * Reopen an existing RocksDB without the delete-and-recreate fallback. Use
this for recovery from
+ * transient errors.
+ */
+ public static org.rocksdb.RocksDB reopenRocksDB(File dbFile) throws
IOException {
+ if (dbFile == null || !dbFile.exists()) {
+ throw new IOException("RocksDB path does not exist: " + dbFile);
+ }
+ Options dbOptions = createDBOptions();
+ try {
+ return org.rocksdb.RocksDB.open(dbOptions, dbFile.toString());
+ } catch (RocksDBException e) {
+ dbOptions.close();
Review Comment:
`reopenRocksDB` allocates native RocksDB resources via `createDBOptions()`
(Options + table config + bloom filter + logger), but on the success path those
handles are never closed. Since auto-recovery can call `reopenRocksDB`
repeatedly, this risks leaking native memory over time. Consider moving Options
(and any dependent native objects) into the `RocksDB` wrapper so they can be
closed on `RocksDB.close()` and also when replacing the instance during
recovery, or otherwise provide an ownership/cleanup mechanism for the success
path.
##########
worker/src/test/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDBRecoverySuiteJ.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.celeborn.service.deploy.worker.shuffledb;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.util.JavaUtils;
+import org.apache.celeborn.service.deploy.worker.WorkerSource;
+
+public class RocksDBRecoverySuiteJ {
+
+ private File dbDir;
+ private File dbFile;
+ private CelebornConf defaultConf;
+ private CelebornConf confWithRecovery;
+ private WorkerSource workerSource;
+ private StoreVersion version;
+
+ @Before
+ public void setUp() throws IOException {
+ dbDir = Files.createTempDirectory("rocksdb-recovery-test").toFile();
+ dbFile = new File(dbDir, "test-db");
+ defaultConf = new CelebornConf();
+ confWithRecovery = new CelebornConf();
+ confWithRecovery.set("celeborn.metadata.autoRecovery.enabled", "true");
+ workerSource = new WorkerSource(defaultConf);
+ version = new StoreVersion(1, 0);
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ workerSource.destroy();
+ JavaUtils.deleteRecursively(dbDir);
+ }
+
+ @Test
+ public void testRecoveryAfterCorruption() throws Exception {
+ DB db = DBProvider.initDB(DBBackend.ROCKSDB, dbFile, version,
workerSource, defaultConf);
+ assertNotNull(db);
+
+ byte[] key = "test-key".getBytes(StandardCharsets.UTF_8);
+ byte[] value = "test-value".getBytes(StandardCharsets.UTF_8);
+ db.put(key, value);
+
+ byte[] result = db.get(key);
+ assertNotNull(result);
+ assertEquals("test-value", new String(result, StandardCharsets.UTF_8));
+
+ db.close();
+
+ // Corrupt the DB by overwriting SST files
+ corruptDbFiles(dbFile);
+
+ // Reopen — initRockDB will wipe and recreate since files are corrupt
+ db = DBProvider.initDB(DBBackend.ROCKSDB, dbFile, version, workerSource,
defaultConf);
+ assertNotNull(db);
+
+ // Data is gone after wipe-and-recreate, but DB is functional
+ byte[] newKey = "new-key".getBytes(StandardCharsets.UTF_8);
+ byte[] newValue = "new-value".getBytes(StandardCharsets.UTF_8);
+ db.put(newKey, newValue);
+
+ result = db.get(newKey);
+ assertNotNull(result);
+ assertEquals("new-value", new String(result, StandardCharsets.UTF_8));
+ db.close();
+ }
+
+ @Test
+ public void testConcurrentRecoveryOnlyRecreatesOnce() throws Exception {
Review Comment:
Test name says "Recreates" but the implementation is a safe reopen of the
same DB path (no delete-and-recreate). Renaming this test to reflect
reopen/recovery behavior would make intent clearer and avoid confusion with the
(destructive) `initRockDB` recreate path.
##########
worker/src/test/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDBRecoverySuiteJ.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.celeborn.service.deploy.worker.shuffledb;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.util.JavaUtils;
+import org.apache.celeborn.service.deploy.worker.WorkerSource;
+
+public class RocksDBRecoverySuiteJ {
+
+ private File dbDir;
+ private File dbFile;
+ private CelebornConf defaultConf;
+ private CelebornConf confWithRecovery;
+ private WorkerSource workerSource;
+ private StoreVersion version;
+
+ @Before
+ public void setUp() throws IOException {
+ dbDir = Files.createTempDirectory("rocksdb-recovery-test").toFile();
+ dbFile = new File(dbDir, "test-db");
+ defaultConf = new CelebornConf();
+ confWithRecovery = new CelebornConf();
+ confWithRecovery.set("celeborn.metadata.autoRecovery.enabled", "true");
+ workerSource = new WorkerSource(defaultConf);
+ version = new StoreVersion(1, 0);
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ workerSource.destroy();
+ JavaUtils.deleteRecursively(dbDir);
+ }
+
+ @Test
+ public void testRecoveryAfterCorruption() throws Exception {
+ DB db = DBProvider.initDB(DBBackend.ROCKSDB, dbFile, version,
workerSource, defaultConf);
+ assertNotNull(db);
+
+ byte[] key = "test-key".getBytes(StandardCharsets.UTF_8);
+ byte[] value = "test-value".getBytes(StandardCharsets.UTF_8);
+ db.put(key, value);
+
+ byte[] result = db.get(key);
+ assertNotNull(result);
+ assertEquals("test-value", new String(result, StandardCharsets.UTF_8));
+
+ db.close();
+
+ // Corrupt the DB by overwriting SST files
+ corruptDbFiles(dbFile);
+
+ // Reopen — initRockDB will wipe and recreate since files are corrupt
+ db = DBProvider.initDB(DBBackend.ROCKSDB, dbFile, version, workerSource,
defaultConf);
+ assertNotNull(db);
+
+ // Data is gone after wipe-and-recreate, but DB is functional
+ byte[] newKey = "new-key".getBytes(StandardCharsets.UTF_8);
+ byte[] newValue = "new-value".getBytes(StandardCharsets.UTF_8);
+ db.put(newKey, newValue);
+
+ result = db.get(newKey);
+ assertNotNull(result);
+ assertEquals("new-value", new String(result, StandardCharsets.UTF_8));
+ db.close();
+ }
+
+ @Test
+ public void testConcurrentRecoveryOnlyRecreatesOnce() throws Exception {
+ DB db = DBProvider.initDB(DBBackend.ROCKSDB, dbFile, version,
workerSource, confWithRecovery);
+ assertNotNull(db);
+
+ byte[] key = "key".getBytes(StandardCharsets.UTF_8);
+ byte[] value = "value".getBytes(StandardCharsets.UTF_8);
+ db.put(key, value);
+
+ RocksDB rocksDB = (RocksDB) db;
+ assertEquals(0, rocksDB.getDbGeneration());
+
+ int threadCount = 8;
+ CyclicBarrier barrier = new CyclicBarrier(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ AtomicInteger recoveryCount = new AtomicInteger(0);
+
Review Comment:
`recoveryCount` is incremented by all threads but never asserted or
otherwise used, so it adds noise without validating behavior. Either remove it,
or assert an expected value (e.g., that all threads executed the code path, or
that only one thread performed the actual recovery depending on what you want
to verify).
##########
docs/configuration/worker.md:
##########
@@ -106,6 +106,7 @@ license: |
| celeborn.worker.graceful.shutdown.enabled | false | false | When true,
during worker shutdown, the worker will wait for all released slots to be
committed or destroyed. | 0.2.0 | |
| celeborn.worker.graceful.shutdown.partitionSorter.shutdownTimeout | 120s |
false | The wait time of waiting for sorting partition files during worker
graceful shutdown. | 0.2.0 | |
| celeborn.worker.graceful.shutdown.recoverDbBackend | ROCKSDB | false |
Specifies a disk-based store used in local db. ROCKSDB or LEVELDB (deprecated).
| 0.4.0 | |
+| celeborn.metadata.autoRecovery.enabled | false | false | If true, the
metadata DB will automatically attempt to recover from RocksDBException errors
during put/get/delete operations. Recovery first tries a safe reopen; if that
fails, it recreates the DB. If false, RocksDBException errors are propagated
directly to the caller. | 0.7.0 | |
Review Comment:
The config description says recovery will "recreate the DB" if safe reopen
fails, but the current implementation in `RocksDB` only attempts a safe reopen
and otherwise propagates the exception (no destructive recreate). Please update
this doc string to match actual behavior to avoid misleading operators.
--
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]
