diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/HBaseTablesProvider.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/HBaseTablesProvider.kt
new file mode 100644
index 00000000..02b3d0bf
--- /dev/null
+++ b/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>
+}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageBackend.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageBackend.kt
index 3cad7ba7..92368622 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageBackend.kt
+++ b/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()
     }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageTable.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageTable.kt
index 79d6d4fb..901d27c2 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/HBaseStorageTable.kt
+++ b/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> {
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/MockHBaseStorageBackend.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/MockHBaseStorageBackend.kt
index d0168823..ff7c8bfc 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/storage/hbase/MockHBaseStorageBackend.kt
+++ b/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
     }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt
index fc1dae98..dfbf5f09 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt
+++ b/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,
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/GraphDefaults.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/GraphDefaults.kt
index 9552485a..84c29641 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/GraphDefaults.kt
+++ b/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
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/compat/DefaultHBaseCluster.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/compat/DefaultHBaseCluster.kt
deleted file mode 100644
index 06f0dd50..00000000
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/compat/DefaultHBaseCluster.kt
+++ /dev/null
@@ -1,224 +0,0 @@
-package com.kakao.actionbase.v2.engine.compat
-
-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 java.lang.AutoCloseable
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.hbase.HBaseConfiguration
-import org.apache.hadoop.hbase.TableName
-import org.apache.hadoop.hbase.client.AsyncConnection
-import org.apache.hadoop.hbase.client.ConnectionFactory
-import org.apache.hadoop.hbase.client.mock.MockHTable
-import org.apache.hadoop.security.UserGroupInformation
-import org.slf4j.LoggerFactory
-
-import reactor.core.publisher.Mono
-import reactor.core.scheduler.Schedulers
-
-/**
- * The original v2 engine was designed to handle multiple HBase clusters.
- * However, in practice, most cases use only one cluster per tenant.
- * Especially for HBase clusters using Kerberos,
- * only one Kerberos principal can connect,
- * so DedicateHBaseCluster was added to handle this situation.
- */
-class DefaultHBaseCluster private constructor(
-    val mock: Boolean,
-    val connectionMono: Mono<AsyncConnection>,
-    val namespace: String,
-    val config: org.apache.hadoop.conf.Configuration,
-) : AutoCloseable {
-    fun getTable(
-        namespace: String,
-        tableName: String,
-    ): Mono<HBaseTables> =
-        if (mock) {
-            val conn = HBaseConnections.getMockConnection(namespace)
-            val table = NewMockTable(conn.getTable(TableName.valueOf("edges")) as MockHTable)
-            val hbaseTable = HBaseTable.create(table)
-            Mono.just(HBaseTables(hbaseTable, hbaseTable))
-        } else {
-            connectionMono.map { conn ->
-                val table = conn.getTable(TableName.valueOf(namespace, tableName))
-                val hbaseTable = HBaseTable.create(table)
-                HBaseTables(hbaseTable, hbaseTable)
-            }
-        }
-
-    // URI format: datastore://{namespace}/{tableName}
-    fun getTable(uri: String): Mono<HBaseTables> {
-        val (namespace, tableName) = parseDatastoreUri(uri)
-        return getTable(namespace, tableName)
-    }
-
-    private fun parseDatastoreUri(uri: String): Pair<String, String> {
-        val parts = uri.removePrefix("datastore://").split("/")
-        require(parts.size == 2) { "Invalid datastore URI: $uri. Expected format: datastore://{namespace}/{tableName}" }
-        return parts[0] to parts[1]
-    }
-
-    override fun close() {
-        connectionMono.block()?.close()
-    }
-
-    companion object {
-        const val DEFAULT_HBASE_NAMESPACE = "default"
-        const val DEFAULT_HBASE_CLUSTER_NAME = "__DEFAULT_HBASE_CLUSTER__"
-        const val LEGACY_DEFAULT_KERBEROS_REALM = "KAKAO.HADOOP"
-
-        private val logger = LoggerFactory.getLogger(DefaultHBaseCluster::class.java)
-
-        private lateinit var instance0: DefaultHBaseCluster
-
-        /**
-         * # default
-         *   secure: true or false
-         *   version: 2.4 or 2.5
-         *
-         * # for 2.4
-         *   hbase.zookeeper.quorum: host1:2181,host2:2181,host3:2181
-         * # for 2.5
-         *   hbase.client.bootstrap.servers: host1:16000,host2:16000,host3:16000
-         *
-         * # for secure cluster
-         *   kerberos.realm: e.g. EXAMPLE.COM (or env AB_KERBEROS_REALM)
-         *     - If missing, defaults to KAKAO.HADOOP for backward compatibility (deprecated)
-         *   krb5ConfPath: /path/to/krb5.conf (or env AB_KRB5_CONF_PATH)
-         *   keytabPath: e.g. /path/to/hadoop-cdl-write.keytab (or env AB_KEYTAB_PATH)
-         *   principal: e.g. hadoop-cdl-write@EXAMPLE.COM (or env AB_PRINCIPAL)
-         */
-        fun initialize(properties: Map<String, String>) {
-            logger.info("KerberosHelper is being initialized.")
-
-            val config = HBaseConfiguration.create()
-
-            if (properties.isEmpty() || properties["version"] == "embedded") {
-                logger.info("🚀 - Using Embedded Mock HBase cluster")
-                instance0 =
-                    DefaultHBaseCluster(
-                        mock = true,
-                        connectionMono = Mono.empty(),
-                        namespace = DEFAULT_HBASE_NAMESPACE,
-                        config = config,
-                    )
-                return
-            }
-
-            val isSecure = properties["secure"]?.toBoolean() ?: false
-            val version = properties["version"] ?: "2.4"
-            val namespace = properties["namespace"] ?: throw IllegalArgumentException("HBase namespace is not set")
-
-            require(version.startsWith("2.4") || version.startsWith("2.5")) {
-                "Unsupported HBase version: $version. Supported versions are 2.4.x and 2.5.x."
-            }
-
-            val krb5ConfPathOpt: String? = properties["krb5ConfPath"] ?: System.getenv("AB_KRB5_CONF_PATH")
-            val principalOpt: String? = properties["principal"] ?: System.getenv("AB_PRINCIPAL")
-            val keytabPathOpt: String? = properties["keytabPath"] ?: System.getenv("AB_KEYTAB_PATH")
-
-            val zookeeperQuorumOpt: String? = properties["hbase.zookeeper.quorum"]
-            val clientBootstrapServersOpt: String? = properties["hbase.client.bootstrap.servers"]
-
-            if (isSecure) {
-                val krb5ConfPath = krb5ConfPathOpt ?: throw IllegalStateException("Kerberos krb5.conf path is not set")
-                val principal = principalOpt ?: throw IllegalStateException("Kerberos principal is not set")
-                val keytabPath = keytabPathOpt ?: throw IllegalStateException("Kerberos keytab path is not set")
-                val kerberosRealm = resolveKerberosRealm(properties)
-
-                System.setProperty("java.security.krb5.conf", krb5ConfPath)
-
-                config["hadoop.security.authentication"] = "kerberos"
-                config["hbase.security.authentication"] = "kerberos"
-                config["hbase.master.kerberos.principal"] = "hbase/_HOST@$kerberosRealm"
-                config["hbase.regionserver.kerberos.principal"] = "hbase/_HOST@$kerberosRealm"
-
-                config["hbase.client.keytab.principal"] = principal
-                config["hbase.client.keytab.file"] = keytabPath
-            }
-
-            if (version.startsWith("2.4")) {
-                logger.info("🚀 - Using HBase 2.4 - zookeeperQuorum: $zookeeperQuorumOpt")
-                config["hbase.zookeeper.quorum"] = zookeeperQuorumOpt ?: throw IllegalStateException("zookeeper.quorum is not set")
-            } else if (version.startsWith("2.5")) {
-                logger.info("🚀 - Using HBase 2.5 - clientBootstrapServers: $clientBootstrapServersOpt")
-                config["hbase.client.registry.impl"] = "org.apache.hadoop.hbase.client.RpcConnectionRegistry"
-                config["hbase.client.bootstrap.servers"] = clientBootstrapServersOpt ?: throw IllegalStateException("hbase.client.bootstrap.servers is not set")
-            } else {
-                throw IllegalArgumentException("Unsupported HBase version: $version. Supported versions are 2.4.x and 2.5.x.")
-            }
-
-            properties.forEach { (key, value) ->
-                if (key.startsWith("hbase.")) {
-                    config[key] = value
-                } else if (key.startsWith("hadoop.")) {
-                    config[key] = value
-                }
-            }
-
-            if (isSecure) {
-                logger.info("🚀 - Using secure HBase cluster with Kerberos authentication")
-                UserGroupInformation.setConfiguration(config)
-            }
-
-            val checkConnectionConfig =
-                org.apache.hadoop.conf
-                    .Configuration(config)
-            // For HBase 2.4.x
-            checkConnectionConfig.setInt("zookeeper.recovery.retry", 1) // HBase 2.4 only
-            checkConnectionConfig.setInt("hbase.client.retries.number", 1) // Common
-
-            // For HBase 2.5+
-            checkConnectionConfig.setInt("hbase.client.connection.registry.impl.retry", 1)
-            checkConnectionConfig.setInt("hbase.client.registry.timeout", 10000)
-            checkConnectionConfig.setInt("hbase.client.operation.timeout", 10000)
-            checkConnectionConfig.setInt("hbase.rpc.timeout", 10000)
-
-            val connectionMono =
-                Mono
-                    .fromFuture(ConnectionFactory.createAsyncConnection(checkConnectionConfig))
-                    .publishOn(Schedulers.boundedElastic())
-                    .doOnSuccess { conn ->
-                        logger.info("🚀 - Successfully established a new HBase connection")
-                        conn.close()
-                    }.flatMap {
-                        Mono.fromFuture(ConnectionFactory.createAsyncConnection(config))
-                    }.cache()
-
-            initialize(connectionMono, namespace, config)
-        }
-
-        internal fun resolveKerberosRealm(
-            properties: Map<String, String>,
-            envKerberosRealm: String? = System.getenv("AB_KERBEROS_REALM"),
-        ): String {
-            val kerberosRealm = (properties["kerberos.realm"] ?: envKerberosRealm)?.trim()
-
-            if (kerberosRealm == null) {
-                logger.warn(
-                    "`kerberos.realm` is not set; falling back to legacy default realm `{}` for backward compatibility. This fallback is deprecated and will be removed in a future release.",
-                    LEGACY_DEFAULT_KERBEROS_REALM,
-                )
-                // TODO(ab#180): Remove legacy fallback and require explicit kerberos.realm after migration period.
-                return LEGACY_DEFAULT_KERBEROS_REALM
-            }
-
-            require(kerberosRealm.isNotEmpty()) { "Kerberos realm must not be blank" }
-            return kerberosRealm
-        }
-
-        fun initialize(
-            connectionMono: Mono<AsyncConnection>,
-            namespace: String,
-            configuration: Configuration,
-        ) {
-            instance0 = DefaultHBaseCluster(mock = false, connectionMono, namespace, configuration)
-        }
-
-        val INSTANCE: DefaultHBaseCluster
-            get() = instance0
-    }
-}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt
index 4fa10b50..fcd14c98 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt
@@ -13,8 +13,8 @@ import com.kakao.actionbase.v2.core.types.EdgeSchema
 import com.kakao.actionbase.v2.engine.GraphDefaults
 import com.kakao.actionbase.v2.engine.edge.HashEdge
 import com.kakao.actionbase.v2.engine.entity.deprecated.DeprecatedEdgeSchema
-import com.kakao.actionbase.v2.engine.label.DatastoreHashLabel
-import com.kakao.actionbase.v2.engine.label.DatastoreIndexedLabel
+import com.kakao.actionbase.v2.engine.label.HBaseStorageBackendHashLabel
+import com.kakao.actionbase.v2.engine.label.HBaseStorageBackendIndexedLabel
 import com.kakao.actionbase.v2.engine.label.Label
 import com.kakao.actionbase.v2.engine.label.hbase.HBaseHashLabel
 import com.kakao.actionbase.v2.engine.label.hbase.HBaseIndexedLabel
@@ -78,7 +78,7 @@ data class LabelEntity(
                         is LocalStorage -> LocalBackedJdbcHashLabel.create(this, graph, storage, block)
                         is JdbcStorage -> JdbcHashLabel.create(this, graph, storage, block)
                         is HBaseStorage -> HBaseHashLabel.create(this, graph, storage)
-                        is DatastoreStorage -> DatastoreHashLabel.create(this, graph, block)
+                        is DatastoreStorage -> HBaseStorageBackendHashLabel.create(this, graph, block)
                         else -> {
                             logger.error(
                                 "{} supports only Local, Jdbc, HBase storage types. {} is not supported. Fallback to NilLabel",
@@ -98,7 +98,7 @@ data class LabelEntity(
 
                     when (storage) {
                         is HBaseStorage -> HBaseIndexedLabel.create(this, graph, storage)
-                        is DatastoreStorage -> DatastoreIndexedLabel.create(this, graph, block)
+                        is DatastoreStorage -> HBaseStorageBackendIndexedLabel.create(this, graph, block)
                         else -> {
                             logger.error(
                                 "{} supports only Jdbc, HBase storage types. {} is not supported. Fallback to NilLabel",
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/DatastoreHashLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/DatastoreHashLabel.kt
deleted file mode 100644
index ca55f40b..00000000
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/DatastoreHashLabel.kt
+++ /dev/null
@@ -1,30 +0,0 @@
-package com.kakao.actionbase.v2.engine.label
-
-import com.kakao.actionbase.v2.core.code.EdgeEncoder
-import com.kakao.actionbase.v2.engine.GraphDefaults
-import com.kakao.actionbase.v2.engine.entity.LabelEntity
-import com.kakao.actionbase.v2.engine.label.hbase.HBaseHashLabel
-import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
-
-import reactor.core.publisher.Mono
-
-class DatastoreHashLabel(
-    entity: LabelEntity,
-    coder: EdgeEncoder<ByteArray>,
-    tables: Mono<HBaseTables>,
-) : HBaseHashLabel(entity, coder, tables) {
-    companion object {
-        fun create(
-            entity: LabelEntity,
-            graph: GraphDefaults,
-            initialize: DatastoreHashLabel.() -> Unit,
-        ): DatastoreHashLabel {
-            val tables = graph.datastore.getTable(entity.storage).cache()
-            return DatastoreHashLabel(
-                entity = entity,
-                coder = graph.edgeEncoderFactory.bytesKeyValueEncoder,
-                tables = tables,
-            ).apply(initialize)
-        }
-    }
-}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/HBaseStorageBackendHashLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/HBaseStorageBackendHashLabel.kt
new file mode 100644
index 00000000..8fcda1cd
--- /dev/null
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/HBaseStorageBackendHashLabel.kt
@@ -0,0 +1,38 @@
+package com.kakao.actionbase.v2.engine.label
+
+import com.kakao.actionbase.engine.storage.DatastoreUri
+import com.kakao.actionbase.engine.storage.DefaultStorageBackendFactory
+import com.kakao.actionbase.engine.storage.HBaseTablesProvider
+import com.kakao.actionbase.v2.core.code.EdgeEncoder
+import com.kakao.actionbase.v2.engine.GraphDefaults
+import com.kakao.actionbase.v2.engine.entity.LabelEntity
+import com.kakao.actionbase.v2.engine.label.hbase.HBaseHashLabel
+import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
+
+import reactor.core.publisher.Mono
+
+class HBaseStorageBackendHashLabel(
+    entity: LabelEntity,
+    coder: EdgeEncoder<ByteArray>,
+    tables: Mono<HBaseTables>,
+) : HBaseHashLabel(entity, coder, tables) {
+    companion object {
+        fun create(
+            entity: LabelEntity,
+            graph: GraphDefaults,
+            initialize: HBaseStorageBackendHashLabel.() -> Unit,
+        ): HBaseStorageBackendHashLabel {
+            val (ns, name) = DatastoreUri.parse(entity.storage)
+            val effectiveNs = ns.ifEmpty { DefaultStorageBackendFactory.defaultNamespace }
+            val provider =
+                DefaultStorageBackendFactory.INSTANCE as? HBaseTablesProvider
+                    ?: throw IllegalStateException("StorageBackend does not support HBaseTables")
+            val tables = provider.getHBaseTables(effectiveNs, name).cache()
+            return HBaseStorageBackendHashLabel(
+                entity = entity,
+                coder = graph.edgeEncoderFactory.bytesKeyValueEncoder,
+                tables = tables,
+            ).apply(initialize)
+        }
+    }
+}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/DatastoreIndexedLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/HBaseStorageBackendIndexedLabel.kt
similarity index 58%
rename from engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/DatastoreIndexedLabel.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/HBaseStorageBackendIndexedLabel.kt
index fdac85b3..ea1df62d 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/DatastoreIndexedLabel.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/HBaseStorageBackendIndexedLabel.kt
@@ -1,5 +1,8 @@
 package com.kakao.actionbase.v2.engine.label
 
+import com.kakao.actionbase.engine.storage.DatastoreUri
+import com.kakao.actionbase.engine.storage.DefaultStorageBackendFactory
+import com.kakao.actionbase.engine.storage.HBaseTablesProvider
 import com.kakao.actionbase.v2.core.code.EdgeEncoder
 import com.kakao.actionbase.v2.core.code.Index
 import com.kakao.actionbase.v2.engine.GraphDefaults
@@ -9,7 +12,7 @@ import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
 
 import reactor.core.publisher.Mono
 
-class DatastoreIndexedLabel(
+class HBaseStorageBackendIndexedLabel(
     entity: LabelEntity,
     coder: EdgeEncoder<ByteArray>,
     indices: List<Index>,
@@ -20,12 +23,17 @@ class DatastoreIndexedLabel(
         fun create(
             entity: LabelEntity,
             graph: GraphDefaults,
-            initialize: DatastoreIndexedLabel.() -> Unit,
-        ): DatastoreIndexedLabel {
+            initialize: HBaseStorageBackendIndexedLabel.() -> Unit,
+        ): HBaseStorageBackendIndexedLabel {
             val indices = entity.indices
             val indexNameToIndex = indices.associateBy { it.name }
-            val tables = graph.datastore.getTable(entity.storage).cache()
-            return DatastoreIndexedLabel(
+            val (ns, name) = DatastoreUri.parse(entity.storage)
+            val effectiveNs = ns.ifEmpty { DefaultStorageBackendFactory.defaultNamespace }
+            val provider =
+                DefaultStorageBackendFactory.INSTANCE as? HBaseTablesProvider
+                    ?: throw IllegalStateException("StorageBackend does not support HBaseTables")
+            val tables = provider.getHBaseTables(effectiveNs, name).cache()
+            return HBaseStorageBackendIndexedLabel(
                 entity = entity,
                 coder = graph.edgeEncoderFactory.bytesKeyValueEncoder,
                 indices = indices,
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseHashLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseHashLabel.kt
index 0c44f9c9..c78bcd66 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseHashLabel.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseHashLabel.kt
@@ -90,7 +90,7 @@ open class HBaseHashLabel(
         return Mono.just(listOf(delete))
     }
 
-    override fun handleDeferredRequests(deferredRequests: List<Any>): Mono<Boolean> = tables.flatMap { it.edge.batch(deferredRequests) }.thenReturn(true)
+    override fun handleDeferredRequests(deferredRequests: List<Any>): Mono<Boolean> = tables.flatMap { it.edge.batchAll(deferredRequests) }.thenReturn(true)
 
     override fun setnx(
         keyField: EncodedKey<ByteArray>,
@@ -228,7 +228,7 @@ open class HBaseHashLabel(
 
         val rows =
             tables
-                .flatMap { it.edge.get(gets) }
+                .flatMap { it.edge.getAll(gets) }
                 .mapNotNull { results ->
                     results
                         .map {
@@ -282,7 +282,7 @@ open class HBaseHashLabel(
 
         val rows =
             tables
-                .flatMap { it.edge.get(gets) }
+                .flatMap { it.edge.getAll(gets) }
                 .mapNotNull { results ->
                     results
                         .map {
@@ -319,7 +319,7 @@ open class HBaseHashLabel(
     fun getActiveStates(gets: List<Get>): Mono<DataFrame> {
         val rows =
             tables
-                .flatMap { it.edge.get(gets) }
+                .flatMap { it.edge.getAll(gets) }
                 .mapNotNull { results ->
                     results
                         .map {
@@ -353,7 +353,7 @@ open class HBaseHashLabel(
                 ).addColumn(Constants.DEFAULT_COLUMN_FAMILY, Constants.DEFAULT_QUALIFIER)
             }
         return tables
-            .flatMap { it.edge.get(gets) }
+            .flatMap { it.edge.getAll(gets) }
             .map {
                 srcAndKeys
                     .map { (src, _) -> src }
@@ -485,7 +485,7 @@ open class HBaseHashLabel(
                 )
                 get.setFilter(complexFilter)
             }
-        return tables.flatMap { it.edge.get(gets) }.map {
+        return tables.flatMap { it.edge.getAll(gets) }.map {
             it.flatMap { result ->
                 val cells = result.listCells() ?: return@flatMap emptyList()
                 cells.map { cell ->
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptions.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptions.kt
index f90ed292..f79eefb5 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptions.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptions.kt
@@ -1,11 +1,9 @@
 package com.kakao.actionbase.v2.engine.storage.hbase
 
-import com.kakao.actionbase.v2.engine.compat.DefaultHBaseCluster
-import com.kakao.actionbase.v2.engine.storage.hbase.impl.NewMockTable
+import com.kakao.actionbase.engine.storage.DefaultStorageBackendFactory
+import com.kakao.actionbase.engine.storage.HBaseTablesProvider
 
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.hbase.TableName
-import org.apache.hadoop.hbase.client.mock.MockHTable
 import org.slf4j.LoggerFactory
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties
@@ -13,8 +11,8 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import reactor.core.publisher.Mono
 
 /**
- * This supports only HBase 2.4 or below.
- * To support HBase 2.5, use [com.kakao.actionbase.v2.engine.compat.DefaultHBaseCluster]
+ * HBase storage options for Label configurations.
+ * Uses DefaultStorageBackendFactory for storage backend access.
  */
 @JsonIgnoreProperties(ignoreUnknown = true)
 data class HBaseOptions(
@@ -24,37 +22,25 @@ data class HBaseOptions(
 ) {
     private val logger = LoggerFactory.getLogger(HBaseOptions::class.java)
 
-    private fun useMockConnection(): Boolean = mock || DefaultHBaseCluster.INSTANCE.mock
-
-    // Mock or DefaultHBaseCluster connections is always available.
+    // Connection is always available via DefaultStorageBackendFactory.
     fun checkConnection(): Mono<Boolean> = Mono.just(true)
 
     /**
-     * // DefaultHBaseCluster = DHC
-     * | mock  | DHC.mock ||| connection    | namespace       | note                               |
-     * |-------|----------|||---------------|-----------------|------------------------------------|
-     * | true  | -        ||| mock          | given namespace | original logic                     |
-     * | -     | true     ||| mock          | given namespace | original logic                     |
-     * | false | false    ||| DHC with      | given namespace | use already created DHC connection |
-     * |-------|----------|||---------------|-----------------|------------------------------------|
+     * Returns the effective namespace, using DefaultStorageBackendFactory's defaultNamespace as fallback.
      */
-    fun getTables(): Mono<HBaseTables> =
-        if (useMockConnection()) {
-            logger.info("Using MockHBase for tableName: {}", tableName)
-            val conn = HBaseConnections.getMockConnection(namespace)
-            val table = NewMockTable(conn.getTable(TableName.valueOf("edges")) as MockHTable)
-            val hbaseTable = HBaseTable.create(table)
-            Mono.just(HBaseTables(hbaseTable, hbaseTable))
-        } else {
-            val namespace = if (namespace.isBlank()) DefaultHBaseCluster.INSTANCE.namespace else this.namespace
-            logger.info("🚀 Using DefaultHBaseCluster for tableName: {} (using namespace: {})", tableName, namespace)
-            DefaultHBaseCluster.INSTANCE.connectionMono
-                .map { connection ->
-                    val edgeTable = connection.getTable(TableName.valueOf(namespace, tableName))
-                    val hbaseTable = HBaseTable.create(edgeTable)
-                    HBaseTables(hbaseTable, hbaseTable)
-                }.cache()
-        }
+    private fun getEffectiveNamespace(): String = namespace.ifEmpty { DefaultStorageBackendFactory.defaultNamespace }
+
+    /**
+     * Returns HBaseTables for Label implementations that need direct HBase table access.
+     */
+    fun getTables(): Mono<HBaseTables> {
+        val effectiveNs = getEffectiveNamespace()
+        logger.debug("Using StorageBackend (HBaseTables) for tableName: {}", tableName)
+        val provider =
+            DefaultStorageBackendFactory.INSTANCE as? HBaseTablesProvider
+                ?: throw IllegalStateException("StorageBackend does not support HBaseTables")
+        return provider.getHBaseTables(effectiveNs, tableName).cache()
+    }
 
     companion object {
         fun newConfiguration(): Configuration = Configuration()
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseTable.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseTable.kt
index f8b2a521..5492b58a 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseTable.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseTable.kt
@@ -29,13 +29,13 @@ interface HBaseTable {
 
     fun get(get: Get): Mono<Result>
 
-    fun get(gets: List<Get>): Mono<List<Result>>
+    fun getAll(gets: List<Get>): Mono<List<Result>>
 
     fun put(put: Put): Mono<Void>
 
     fun delete(delete: Delete): Mono<Void>
 
-    fun batch(deferredRequests: List<Any>): Mono<Void>
+    fun batchAll(deferredRequests: List<Any>): Mono<Void>
 
     fun exists(get: Get): Mono<Boolean>
 
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseAsyncTable.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseAsyncTable.kt
index 5c0a76c8..1ec840b2 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseAsyncTable.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseAsyncTable.kt
@@ -33,7 +33,7 @@ class HBaseAsyncTable(
 
     override fun get(get: Get): Mono<Result> = Mono.fromFuture(asyncTable.get(get))
 
-    override fun get(gets: List<Get>): Mono<List<Result>> {
+    override fun getAll(gets: List<Get>): Mono<List<Result>> {
         val futures = asyncTable.getAll(gets)
         return Mono.fromFuture(futures)
     }
@@ -42,7 +42,7 @@ class HBaseAsyncTable(
 
     override fun delete(delete: Delete): Mono<Void> = Mono.fromFuture(asyncTable.delete(delete))
 
-    override fun batch(deferredRequests: List<Any>): Mono<Void> {
+    override fun batchAll(deferredRequests: List<Any>): Mono<Void> {
         val mutations: List<Mutation> =
             deferredRequests.map {
                 when (it) {
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseSyncTable.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseSyncTable.kt
index ae00ec1f..834c3007 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseSyncTable.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/impl/HBaseSyncTable.kt
@@ -36,13 +36,13 @@ class HBaseSyncTable(
 
     override fun get(get: Get): Mono<Result> = Mono.fromCallable { table.get(get) }
 
-    override fun get(gets: List<Get>): Mono<List<Result>> = Mono.fromCallable { table.get(gets).asList() }
+    override fun getAll(gets: List<Get>): Mono<List<Result>> = Mono.fromCallable { table.get(gets).asList() }
 
     override fun put(put: Put): Mono<Void> = Mono.fromCallable { table.put(put) }.then()
 
     override fun delete(delete: Delete): Mono<Void> = Mono.fromCallable { table.delete(delete) }.then()
 
-    override fun batch(deferredRequests: List<Any>): Mono<Void> {
+    override fun batchAll(deferredRequests: List<Any>): Mono<Void> {
         val mutations: List<Mutation> =
             deferredRequests.map {
                 when (it) {
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/compat/DefaultHBaseClusterTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/compat/DefaultHBaseClusterTest.kt
index 8e38b1d7..3da20aab 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/compat/DefaultHBaseClusterTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/compat/DefaultHBaseClusterTest.kt
@@ -1,7 +1,8 @@
 package com.kakao.actionbase.v2.engine.compat
 
+import com.kakao.actionbase.engine.storage.hbase.HBaseStorageBackend
+
 import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 
@@ -19,14 +20,14 @@ class DefaultHBaseClusterTest {
 
     @Test
     fun `missing kerberos realm should use legacy default for compatibility`() {
-        val kerberosRealm = DefaultHBaseCluster.resolveKerberosRealm(secureBaseProperties, null)
+        val kerberosRealm = HBaseStorageBackend.resolveKerberosRealm(secureBaseProperties, null)
 
-        assertEquals(DefaultHBaseCluster.LEGACY_DEFAULT_KERBEROS_REALM, kerberosRealm)
+        assertEquals(HBaseStorageBackend.LEGACY_DEFAULT_KERBEROS_REALM, kerberosRealm)
     }
 
     @Test
     fun `environment kerberos realm should be used when property is missing`() {
-        val kerberosRealm = DefaultHBaseCluster.resolveKerberosRealm(secureBaseProperties, "ENV.EXAMPLE.COM")
+        val kerberosRealm = HBaseStorageBackend.resolveKerberosRealm(secureBaseProperties, "ENV.EXAMPLE.COM")
 
         assertEquals("ENV.EXAMPLE.COM", kerberosRealm)
     }
@@ -35,7 +36,7 @@ class DefaultHBaseClusterTest {
     fun `property kerberos realm should override environment realm`() {
         val properties = secureBaseProperties + ("kerberos.realm" to "PROP.EXAMPLE.COM")
 
-        val kerberosRealm = DefaultHBaseCluster.resolveKerberosRealm(properties, "ENV.EXAMPLE.COM")
+        val kerberosRealm = HBaseStorageBackend.resolveKerberosRealm(properties, "ENV.EXAMPLE.COM")
 
         assertEquals("PROP.EXAMPLE.COM", kerberosRealm)
     }
@@ -44,35 +45,19 @@ class DefaultHBaseClusterTest {
     fun `kerberos realm should be trimmed`() {
         val properties = secureBaseProperties + ("kerberos.realm" to "  EXAMPLE.COM  ")
 
-        val kerberosRealm = DefaultHBaseCluster.resolveKerberosRealm(properties, null)
+        val kerberosRealm = HBaseStorageBackend.resolveKerberosRealm(properties, null)
 
         assertEquals("EXAMPLE.COM", kerberosRealm)
     }
 
     @Test
-    fun `blank kerberos realm should throw for secure cluster`() {
+    fun `blank kerberos realm should throw`() {
         val properties = secureBaseProperties + ("kerberos.realm" to "  ")
 
         val exception =
             assertThrows<IllegalArgumentException> {
-                DefaultHBaseCluster.initialize(properties)
+                HBaseStorageBackend.resolveKerberosRealm(properties, null)
             }
         assertEquals("Kerberos realm must not be blank", exception.message)
     }
-
-    @Test
-    fun `embedded version should skip kerberos configuration`() {
-        val properties = mapOf("version" to "embedded")
-
-        DefaultHBaseCluster.initialize(properties)
-        assertTrue(DefaultHBaseCluster.INSTANCE.mock)
-    }
-
-    @Test
-    fun `empty properties should use mock cluster`() {
-        val properties = emptyMap<String, String>()
-
-        DefaultHBaseCluster.initialize(properties)
-        assertTrue(DefaultHBaseCluster.INSTANCE.mock)
-    }
 }
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptionsTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptionsTest.kt
index f7bb7bc4..994f0bad 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptionsTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/hbase/HBaseOptionsTest.kt
@@ -59,9 +59,7 @@ class HBaseOptionsTest(
     @Test
     fun `use default hbase cluster namespace`() =
         test(config.tableName) { tableName ->
-
-            Storage
-                .parseOptions<HBaseOptions>(makeConfig("", tableName.qualifierAsString))
+            Storage.parseOptions<HBaseOptions>(makeConfig("", tableName.qualifierAsString))
         }
 
     private fun makeConfig(
@@ -87,7 +85,8 @@ class HBaseOptionsTest(
                 assertNotNull(tables)
                 assertNotNull(tables.edge)
                 assertNotNull(tables.lock)
-                assertEquals(expectTableName, tables.edge.name)
+                // MockHBaseStorageBackend uses table name without namespace prefix
+                assertEquals(expectTableName.qualifierAsString, tables.edge.name.qualifierAsString)
             }.verifyComplete()
     }
 
diff --git a/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingClusterExtension.kt b/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingClusterExtension.kt
index 5716dba0..90487734 100644
--- a/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingClusterExtension.kt
+++ b/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingClusterExtension.kt
@@ -1,7 +1,6 @@
 package com.kakao.actionbase.test.hbase
 
 import com.kakao.actionbase.engine.storage.DefaultStorageBackendFactory
-import com.kakao.actionbase.v2.engine.compat.DefaultHBaseCluster
 
 import org.apache.hadoop.hbase.client.AsyncConnection
 import org.apache.hadoop.hbase.client.AsyncTable
@@ -27,7 +26,6 @@ class HBaseTestingClusterExtension :
 
     override fun beforeAll(context: ExtensionContext) {
         HBaseTestingCluster.startIfNeeded()
-        DefaultHBaseCluster.initialize(Mono.just(HBaseTestingCluster.asyncConnection), "ab_test", HBaseTestingCluster.hbaseConfiguration)
         // Initialize DefaultStorageBackendFactory with the HBase testing cluster (idempotent)
         val testingBackend = HBaseTestingStorageBackend(Mono.just(HBaseTestingCluster.asyncConnection), "ab_test")
         DefaultStorageBackendFactory.initialize(testingBackend, "ab_test")
diff --git a/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingStorageBackend.kt b/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingStorageBackend.kt
index 87dd6dd6..02bb833b 100644
--- a/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingStorageBackend.kt
+++ b/engine/src/testFixtures/kotlin/com/kakao/actionbase/test/hbase/HBaseTestingStorageBackend.kt
@@ -1,9 +1,11 @@
 package com.kakao.actionbase.test.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.engine.storage.hbase.HBaseStorageTable
 import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTable
+import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
 
 import org.apache.hadoop.hbase.TableName
 import org.apache.hadoop.hbase.client.AsyncConnection
@@ -17,7 +19,8 @@ import reactor.core.publisher.Mono
 class HBaseTestingStorageBackend(
     private val connectionMono: Mono<AsyncConnection>,
     private val defaultNamespace: String,
-) : StorageBackend {
+) : StorageBackend,
+    HBaseTablesProvider {
     override fun getStorageTable(
         namespace: String,
         name: String,
@@ -31,6 +34,19 @@ class HBaseTestingStorageBackend(
         }
     }
 
+    override fun getHBaseTables(
+        namespace: String,
+        name: String,
+    ): Mono<HBaseTables> {
+        val effectiveNs = namespace.ifEmpty { defaultNamespace }
+        return connectionMono.map { conn ->
+            val tableName = TableName.valueOf(effectiveNs, name)
+            val asyncTable = conn.getTable(tableName)
+            val hbaseTable = HBaseTable.create(asyncTable)
+            HBaseTables(hbaseTable, hbaseTable)
+        }
+    }
+
     override fun close() {
         // Connection is managed by HBaseTestingCluster
     }
diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/HBaseDatastoreBindingConfiguration.kt b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/HBaseDatastoreBindingConfiguration.kt
index c3e02a6e..dc6cde99 100644
--- a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/HBaseDatastoreBindingConfiguration.kt
+++ b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/HBaseDatastoreBindingConfiguration.kt
@@ -1,8 +1,9 @@
 package com.kakao.actionbase.server.configuration
 
 import com.kakao.actionbase.engine.datastore.hbase.admin.HBaseAdmin
+import com.kakao.actionbase.engine.storage.DefaultStorageBackendFactory
+import com.kakao.actionbase.engine.storage.hbase.HBaseStorageBackend
 import com.kakao.actionbase.v2.engine.Graph
-import com.kakao.actionbase.v2.engine.compat.DefaultHBaseCluster
 
 import org.apache.hadoop.hbase.NamespaceDescriptor
 import org.springframework.context.annotation.Bean
@@ -11,16 +12,16 @@ import org.springframework.context.annotation.Configuration
 @Configuration
 @ConditionalOnHBaseDatastore
 class HBaseDatastoreBindingConfiguration(
-    // DefaultHBaseCluster is initialized in graph, so graph configuration must be completed before hbase admin injection is possible.
+    // DefaultStorageBackendFactory is initialized in graph, so graph configuration must be completed before hbase admin injection is possible.
     private val graph: Graph,
 ) {
     @Bean
-    fun hBaseAdmin(): HBaseAdmin =
-        HBaseAdmin(
-            DefaultHBaseCluster.INSTANCE.connectionMono
-                .map { it.admin }
-                .cache(),
-        )
+    fun hBaseAdmin(): HBaseAdmin {
+        val backend =
+            DefaultStorageBackendFactory.INSTANCE as? HBaseStorageBackend
+                ?: throw IllegalStateException("HBaseAdmin requires HBaseStorageBackend but got ${DefaultStorageBackendFactory.INSTANCE::class.simpleName}")
+        return HBaseAdmin(backend.getAdminMono())
+    }
 
     @Bean
     fun namespaceDescriptor(serverProperties: ServerProperties): NamespaceDescriptor =