Replace DefaultHBaseCluster with DefaultStorageBackendFactory

engine/src/main/kotlin/com/kakao/actionbase/engine/storage/HBaseTablesProvider.kt

@@ -0,0 +1,17 @@
+package com.kakao.actionbase.engine.storage
+
+import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
+
+import reactor.core.publisher.Mono
+
+/**
+ * Provides HBaseTables for v2 Label implementations that need direct HBase table access
+ * (e.g., Filters, CellUtil) beyond what StorageTable supports.
+ */
+@Deprecated("backwards compatibility for v2, use StorageBackend instead")
+interface HBaseTablesProvider {
+    fun getHBaseTables(
+        namespace: String,
+        name: String,
+    ): Mono<HBaseTables>
+}

engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageBackend.kt

@@ -1,12 +1,15 @@
 package com.kakao.actionbase.engine.storage.hbase
 
+import com.kakao.actionbase.engine.storage.HBaseTablesProvider
 import com.kakao.actionbase.engine.storage.StorageBackend
 import com.kakao.actionbase.engine.storage.StorageTable
 import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTable
+import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.hbase.HBaseConfiguration
 import org.apache.hadoop.hbase.TableName
+import org.apache.hadoop.hbase.client.AsyncAdmin
 import org.apache.hadoop.hbase.client.AsyncConnection
 import org.apache.hadoop.hbase.client.ConnectionFactory
 import org.apache.hadoop.security.UserGroupInformation
@@ -17,7 +20,8 @@ import reactor.core.scheduler.Schedulers
 
 class HBaseStorageBackend private constructor(
     private val connectionMono: Mono<AsyncConnection>,
-) : StorageBackend {
+) : StorageBackend,
+    HBaseTablesProvider {
     override fun getStorageTable(
         namespace: String,
         name: String,
@@ -28,6 +32,18 @@ class HBaseStorageBackend private constructor(
             HBaseStorageTable(hbaseTable)
         }
 
+    override fun getHBaseTables(
+        namespace: String,
+        name: String,
+    ): Mono<HBaseTables> =
+        connectionMono.map { conn ->
+            val table = conn.getTable(TableName.valueOf(namespace, name))
+            val hbaseTable = HBaseTable.create(table)
+            HBaseTables(hbaseTable, hbaseTable)
+        }
+
+    fun getAdminMono(): Mono<AsyncAdmin> = connectionMono.map { it.admin }.cache()
+
     override fun close() {
         connectionMono.block()?.close()
     }

engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageTable.kt

@@ -18,7 +18,8 @@ import reactor.core.publisher.Mono
 
 class HBaseStorageTable(
     private val table: HBaseTable,
-) : StorageTable {
+) : StorageTable,
+    HBaseTable by table {
     override fun get(key: ByteArray): Mono<ByteArray?> {
         val get = Get(key).addColumn(Constants.DEFAULT_COLUMN_FAMILY, Constants.DEFAULT_QUALIFIER)
         return table.get(get).handle { result, sink ->
@@ -31,7 +32,7 @@ class HBaseStorageTable(
 
     override fun get(keys: List<ByteArray>): Mono<List<HBaseRecord>> {
         val gets = keys.map { Get(it).addColumn(Constants.DEFAULT_COLUMN_FAMILY, Constants.DEFAULT_QUALIFIER) }
-        return table.get(gets).map { results ->
+        return table.getAll(gets).map { results ->
             results.filter { !it.isEmpty }.map { result ->
                 HBaseRecord(
                     key = result.row,
@@ -107,7 +108,7 @@ class HBaseStorageTable(
                         )
                 }
             }
-        return table.batch(mutations)
+        return table.batchAll(mutations)
     }
 
     override fun exists(key: ByteArray): Mono<Boolean> {

engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/MockHBaseStorageBackend.kt

@@ -1,9 +1,11 @@
 package com.kakao.actionbase.engine.storage.hbase
 
+import com.kakao.actionbase.engine.storage.HBaseTablesProvider
 import com.kakao.actionbase.engine.storage.StorageBackend
 import com.kakao.actionbase.engine.storage.StorageTable
 import com.kakao.actionbase.v2.engine.storage.hbase.HBaseConnections
 import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTable
+import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
 import com.kakao.actionbase.v2.engine.storage.hbase.impl.NewMockTable
 
 import org.apache.hadoop.hbase.TableName
@@ -17,7 +19,9 @@ import reactor.core.publisher.Mono
  *
  * Each namespace + name combination gets its own isolated table.
  */
-class MockHBaseStorageBackend : StorageBackend {
+class MockHBaseStorageBackend :
+    StorageBackend,
+    HBaseTablesProvider {
     override fun getStorageTable(
         namespace: String,
         name: String,
@@ -26,6 +30,14 @@ class MockHBaseStorageBackend : StorageBackend {
         return Mono.just(HBaseStorageTable(hbaseTable))
     }
 
+    override fun getHBaseTables(
+        namespace: String,
+        name: String,
+    ): Mono<HBaseTables> {
+        val hbaseTable = createMockHBaseTable(namespace, name)
+        return Mono.just(HBaseTables(hbaseTable, hbaseTable))
+    }
+
     override fun close() {
         // nothing to close
     }

engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt

@@ -1,5 +1,6 @@
 package com.kakao.actionbase.v2.engine
 
+import com.kakao.actionbase.engine.storage.DefaultStorageBackendFactory
 import com.kakao.actionbase.v2.core.code.EdgeEncoderFactory
 import com.kakao.actionbase.v2.core.code.EmptyEdgeIdEncoder
 import com.kakao.actionbase.v2.core.code.IdEdgeEncoder
@@ -16,7 +17,6 @@ import com.kakao.actionbase.v2.engine.cdc.CdcContext
 import com.kakao.actionbase.v2.engine.cdc.CdcFactory
 import com.kakao.actionbase.v2.engine.client.kafka.KafkaClientFactory
 import com.kakao.actionbase.v2.engine.client.web.WebClientFactory
-import com.kakao.actionbase.v2.engine.compat.DefaultHBaseCluster
 import com.kakao.actionbase.v2.engine.edge.MutationResult
 import com.kakao.actionbase.v2.engine.edge.MutationResultItem
 import com.kakao.actionbase.v2.engine.entity.AliasEntity
@@ -92,7 +92,6 @@ class Graph(
     override val metastore: Database,
     override val metadataTable: MetadataTable,
     override val edgeEncoderFactory: EdgeEncoderFactory,
-    override val datastore: DefaultHBaseCluster,
     private val systemStorages: Map<EntityName, StorageEntity>,
     config: GraphConfig,
     serviceLabel: Label,
@@ -892,7 +891,7 @@ class Graph(
         intervalDisposable?.dispose()
         log.info("Disposed Flux.interval for reloading metastore - {}", intervalDisposable)
         HBaseConnections.closeConnections().block()
-        DefaultHBaseCluster.INSTANCE.close()
+        DefaultStorageBackendFactory.close()
     }
 
     fun status(name: EntityName): Mono<String> = getLabel(name).status()
@@ -941,7 +940,8 @@ class Graph(
             kafkaClientFactory: KafkaClientFactory,
             webClientFactory: WebClientFactory,
         ): Graph {
-            DefaultHBaseCluster.initialize(config.hbase)
+            // Initialize storage backend if not already initialized (idempotent)
+            DefaultStorageBackendFactory.initialize(config.hbase)
             log.info("phase: {}", config.phase)
             log.info("tenant: {}", config.tenant)
             log.info("graph config: {}", config)
@@ -999,7 +999,6 @@ class Graph(
                     metadataTable,
                     edgeEncoderFactory,
                     storageEntities,
-                    DefaultHBaseCluster.INSTANCE,
                 )
 
             val serviceLabel =
@@ -1061,7 +1060,6 @@ class Graph(
                 defaults.metastore,
                 defaults.metadataTable,
                 defaults.edgeEncoderFactory,
-                defaults.datastore,
                 defaults.storages,
                 config,
                 serviceLabel,

engine/src/main/kotlin/com/kakao/actionbase/v2/engine/GraphDefaults.kt

@@ -2,7 +2,6 @@ package com.kakao.actionbase.v2.engine
 
 import com.kakao.actionbase.engine.EngineConstants
 import com.kakao.actionbase.v2.core.code.EdgeEncoderFactory
-import com.kakao.actionbase.v2.engine.compat.DefaultHBaseCluster
 import com.kakao.actionbase.v2.engine.entity.EntityName
 import com.kakao.actionbase.v2.engine.entity.StorageEntity
 import com.kakao.actionbase.v2.engine.metadata.StorageType
@@ -16,7 +15,6 @@ interface GraphDefaults {
     val metadataTable: MetadataTable
     val storages: Map<EntityName, StorageEntity>
     val edgeEncoderFactory: EdgeEncoderFactory
-    val datastore: DefaultHBaseCluster
 
     fun getStorage(uri: String): StorageEntity? =
         when {
@@ -35,5 +33,4 @@ data class AbstractGraphDefaults(
     override val metadataTable: MetadataTable,
     override val edgeEncoderFactory: EdgeEncoderFactory,
     override val storages: Map<EntityName, StorageEntity>,
-    override val datastore: DefaultHBaseCluster,
 ) : GraphDefaults