You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bifromq.apache.org by "popduke (via GitHub)" <gi...@apache.org> on 2025/11/07 06:08:16 UTC
[PR] Update merge protocol & various improvements for production stability [bifromq]
popduke opened a new pull request, #188:
URL: https://github.com/apache/bifromq/pull/188
* Base-KV improvements:
Loosens the merge prerequisite on config alignment, enhances the RedundantRangeRemovalBalancer to detect and purge zombie ranges, refactors the local engine for pluggability, and rolls out multiple optimizations—snapshot session reuse, obsolete snapshot filtering, leaner installation flow, lower split/merge memory pressure, faster distributed range lookup, RocksDB internal metrics toggles, redefined WAL compaction thresholds, improved WAL read path, and better default local-engine settings—alongside pluggable split hinters.
* Inbox service optimizations:
Cuts read/write overhead in InboxStoreCoProc and accelerates BatchInsertRequest marshalling.
* Multi-tenancy enhancements:
Adds tenant-level observability metrics and introduces a tenant-level switch to publish will messages during shutdown.
* Bug fixes:
Resolves a NoSuchElementException when parsing AgentMetadata from CRDT and eliminates task duplication in BatchQueryCall.
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "Gujiawei-Edinburgh (via GitHub)" <gi...@apache.org>.
Gujiawei-Edinburgh commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2514239207
##########
build/build-bifromq-starter/src/main/java/org/apache/bifromq/starter/config/StandaloneConfigConsolidator.java:
##########
@@ -179,4 +237,91 @@ private static ServerSSLContextConfig genSelfSignedServerCert() throws Certifica
sslContextConfig.setKeyFile(selfCert.privateKey().getAbsolutePath());
return sslContextConfig;
}
+
+ private static IKVEngineProvider findProvider(String type) {
+ Map<String, IKVEngineProvider> providers = BaseHookLoader.load(IKVEngineProvider.class);
+ IKVEngineProvider found = null;
+ for (IKVEngineProvider p : providers.values()) {
+ if (p.type().equalsIgnoreCase(type)) {
+ if (found != null) {
+ throw new IllegalStateException("Duplicate storage engine provider type: " + type);
Review Comment:
Would it be better to log a warning and ignore the duplicate provider?
##########
build/build-bifromq-starter/src/main/java/org/apache/bifromq/starter/config/StandaloneConfigConsolidator.java:
##########
@@ -179,4 +237,91 @@ private static ServerSSLContextConfig genSelfSignedServerCert() throws Certifica
sslContextConfig.setKeyFile(selfCert.privateKey().getAbsolutePath());
return sslContextConfig;
}
+
+ private static IKVEngineProvider findProvider(String type) {
+ Map<String, IKVEngineProvider> providers = BaseHookLoader.load(IKVEngineProvider.class);
+ IKVEngineProvider found = null;
+ for (IKVEngineProvider p : providers.values()) {
+ if (p.type().equalsIgnoreCase(type)) {
+ if (found != null) {
+ throw new IllegalStateException("Duplicate storage engine provider type: " + type);
Review Comment:
Would it be better to log a warning and ignore the duplicate providers?
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "popduke (via GitHub)" <gi...@apache.org>.
popduke commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2516334004
##########
base-kv/base-kv-local-engine-rocksdb/src/main/java/org/apache/bifromq/basekv/localengine/rocksdb/RocksDBKVSpaceMigratableWriter.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.bifromq.basekv.localengine.rocksdb;
+
+import com.google.protobuf.ByteString;
+import java.util.Map;
+import java.util.function.Consumer;
+import org.apache.bifromq.basekv.localengine.IKVSpaceIterator;
+import org.apache.bifromq.basekv.localengine.IKVSpaceMigratableWriter;
+import org.apache.bifromq.basekv.localengine.IRestoreSession;
+import org.apache.bifromq.basekv.localengine.ISyncContext;
+import org.apache.bifromq.basekv.localengine.KVEngineException;
+import org.apache.bifromq.basekv.localengine.metrics.KVSpaceOpMeters;
+import org.apache.bifromq.basekv.proto.Boundary;
+import org.rocksdb.WriteOptions;
+import org.slf4j.Logger;
+
+class RocksDBKVSpaceMigratableWriter extends RocksDBKVSpaceWriter
+ implements IKVSpaceMigratableWriter {
+
+ RocksDBKVSpaceMigratableWriter(String id,
+ IRocksDBKVSpaceEpochHandle dbHandle,
+ RocksDBKVEngine<?> engine,
+ WriteOptions writeOptions,
+ ISyncContext syncContext,
+ IWriteStatsRecorder.IRecorder writeStatsRecorder,
+ Consumer<Map<ByteString, ByteString>> afterWrite,
+ KVSpaceOpMeters opMeters,
+ Logger logger) {
+ super(id, dbHandle, engine, writeOptions, syncContext, writeStatsRecorder, afterWrite, opMeters, logger);
+ }
+
+ @Override
+ public IRestoreSession migrateTo(String targetSpaceId, Boundary boundary) {
+ try {
+ RocksDBCPableKVSpace targetKVSpace = (RocksDBCPableKVSpace) engine.createIfMissing(targetSpaceId);
+ IRestoreSession targetSpaceRestoreSession = targetKVSpace.startRestore((count, bytes) ->
+ logger.debug("Migrate {} kv to space[{}] from space[{}]: startKey={}, endKey={}",
+ count, targetSpaceId, id, boundary.getStartKey().toStringUtf8(),
+ boundary.getEndKey().toStringUtf8()));
+ try (IKVSpaceIterator itr = new RocksDBKVSpaceIterator(new RocksDBSnapshot(dbHandle, null), boundary,
+ new IteratorOptions(false, 52428))) {
+ for (itr.seekToFirst(); itr.isValid(); itr.next()) {
Review Comment:
data migration is not transactional at this level, the split/merge process will ensure the consistency.
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "Gujiawei-Edinburgh (via GitHub)" <gi...@apache.org>.
Gujiawei-Edinburgh commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2510757482
##########
base-env/base-env-provider/src/main/java/org/apache/bifromq/baseenv/ZeroCopyParser.java:
##########
@@ -43,4 +44,17 @@ public static <T> T parse(ByteString bytes, Parser<T> parser) throws InvalidProt
input.enableAliasing(true);
Review Comment:
From this PR, zero-copy parser is used more widely. How about the case of direct buffer memory usage? And will it cause OOM when the parsing is extremly frequent?
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "Gujiawei-Edinburgh (via GitHub)" <gi...@apache.org>.
Gujiawei-Edinburgh commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2510757482
##########
base-env/base-env-provider/src/main/java/org/apache/bifromq/baseenv/ZeroCopyParser.java:
##########
@@ -43,4 +44,17 @@ public static <T> T parse(ByteString bytes, Parser<T> parser) throws InvalidProt
input.enableAliasing(true);
Review Comment:
From this PR, zero-copy parser is used more widely. How about the case of direct buffer memory usage?
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "Gujiawei-Edinburgh (via GitHub)" <gi...@apache.org>.
Gujiawei-Edinburgh commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2510757482
##########
base-env/base-env-provider/src/main/java/org/apache/bifromq/baseenv/ZeroCopyParser.java:
##########
@@ -43,4 +44,17 @@ public static <T> T parse(ByteString bytes, Parser<T> parser) throws InvalidProt
input.enableAliasing(true);
Review Comment:
From this PR, zero-copy parser is used more widely. How about the case of direct buffer memory usage? And will it cause OOM when the parsing is extremly frequently?
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "Gujiawei-Edinburgh (via GitHub)" <gi...@apache.org>.
Gujiawei-Edinburgh commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2514252837
##########
base-env/base-env-provider/src/main/java/org/apache/bifromq/baseenv/ZeroCopyParser.java:
##########
@@ -43,4 +44,17 @@ public static <T> T parse(ByteString bytes, Parser<T> parser) throws InvalidProt
input.enableAliasing(true);
Review Comment:
I am not pretty sure, since from the enableAliasing comments, it says: "Enables ByteString aliasing of the underlying buffer, trading off on buffer pinning for data copies. Only valid for buffer-backed streams". So i think the pinning buffer here means the direct buffer memory will be pinned until the PB obj is garbage collected.
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "popduke (via GitHub)" <gi...@apache.org>.
popduke commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2512557857
##########
base-env/base-env-provider/src/main/java/org/apache/bifromq/baseenv/ZeroCopyParser.java:
##########
@@ -43,4 +44,17 @@ public static <T> T parse(ByteString bytes, Parser<T> parser) throws InvalidProt
input.enableAliasing(true);
Review Comment:
zero-copy parser is only applied in the context of parsing PB from byte string backed by byte array. For direct buffer, things become way more complex, and it will be considered in future release.
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "popduke (via GitHub)" <gi...@apache.org>.
popduke commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2516325575
##########
build/build-bifromq-starter/src/main/java/org/apache/bifromq/starter/config/StandaloneConfigConsolidator.java:
##########
@@ -179,4 +237,91 @@ private static ServerSSLContextConfig genSelfSignedServerCert() throws Certifica
sslContextConfig.setKeyFile(selfCert.privateKey().getAbsolutePath());
return sslContextConfig;
}
+
+ private static IKVEngineProvider findProvider(String type) {
+ Map<String, IKVEngineProvider> providers = BaseHookLoader.load(IKVEngineProvider.class);
+ IKVEngineProvider found = null;
+ for (IKVEngineProvider p : providers.values()) {
+ if (p.type().equalsIgnoreCase(type)) {
+ if (found != null) {
+ throw new IllegalStateException("Duplicate storage engine provider type: " + type);
Review Comment:
This is by design. It's dangerous to have name-conflict local engine impls running in a bifromq process, the loading order is not guaranteed between restart.
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "Gujiawei-Edinburgh (via GitHub)" <gi...@apache.org>.
Gujiawei-Edinburgh commented on code in PR #188:
URL: https://github.com/apache/bifromq/pull/188#discussion_r2514613239
##########
base-kv/base-kv-local-engine-rocksdb/src/main/java/org/apache/bifromq/basekv/localengine/rocksdb/RocksDBKVSpaceMigratableWriter.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.bifromq.basekv.localengine.rocksdb;
+
+import com.google.protobuf.ByteString;
+import java.util.Map;
+import java.util.function.Consumer;
+import org.apache.bifromq.basekv.localengine.IKVSpaceIterator;
+import org.apache.bifromq.basekv.localengine.IKVSpaceMigratableWriter;
+import org.apache.bifromq.basekv.localengine.IRestoreSession;
+import org.apache.bifromq.basekv.localengine.ISyncContext;
+import org.apache.bifromq.basekv.localengine.KVEngineException;
+import org.apache.bifromq.basekv.localengine.metrics.KVSpaceOpMeters;
+import org.apache.bifromq.basekv.proto.Boundary;
+import org.rocksdb.WriteOptions;
+import org.slf4j.Logger;
+
+class RocksDBKVSpaceMigratableWriter extends RocksDBKVSpaceWriter
+ implements IKVSpaceMigratableWriter {
+
+ RocksDBKVSpaceMigratableWriter(String id,
+ IRocksDBKVSpaceEpochHandle dbHandle,
+ RocksDBKVEngine<?> engine,
+ WriteOptions writeOptions,
+ ISyncContext syncContext,
+ IWriteStatsRecorder.IRecorder writeStatsRecorder,
+ Consumer<Map<ByteString, ByteString>> afterWrite,
+ KVSpaceOpMeters opMeters,
+ Logger logger) {
+ super(id, dbHandle, engine, writeOptions, syncContext, writeStatsRecorder, afterWrite, opMeters, logger);
+ }
+
+ @Override
+ public IRestoreSession migrateTo(String targetSpaceId, Boundary boundary) {
+ try {
+ RocksDBCPableKVSpace targetKVSpace = (RocksDBCPableKVSpace) engine.createIfMissing(targetSpaceId);
+ IRestoreSession targetSpaceRestoreSession = targetKVSpace.startRestore((count, bytes) ->
+ logger.debug("Migrate {} kv to space[{}] from space[{}]: startKey={}, endKey={}",
+ count, targetSpaceId, id, boundary.getStartKey().toStringUtf8(),
+ boundary.getEndKey().toStringUtf8()));
+ try (IKVSpaceIterator itr = new RocksDBKVSpaceIterator(new RocksDBSnapshot(dbHandle, null), boundary,
+ new IteratorOptions(false, 52428))) {
+ for (itr.seekToFirst(); itr.isValid(); itr.next()) {
Review Comment:
During this loop, if an error occurs midway, what happens to the data that has already been flushed in the same loop? Each operation checks `shouldFlush`.
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update merge protocol & various improvements for production stability [bifromq]
Posted by "popduke (via GitHub)" <gi...@apache.org>.
popduke merged PR #188:
URL: https://github.com/apache/bifromq/pull/188
--
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: commits-unsubscribe@bifromq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org