Update merge protocol & various improvements for production stability

.github/workflows/build-cov.yaml

@@ -9,18 +9,43 @@ on:
     branches:
       - 'main'
 jobs:
-  check-license:
-    name: "Check License"
+  prepare:
+    name: "Prepare"
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+      - name: Set up JDK 17
+        uses: actions/setup-java@v4
+        with:
+          java-version: '17'
+          distribution: 'temurin'
+          cache: maven
+      - name: Install without tests
+        run: mvn -q -DskipTests install
+
+  license:
+    name: "License Check"
+    needs: prepare
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up JDK 17
+        uses: actions/setup-java@v4
+        with:
+          java-version: '17'
+          distribution: 'temurin'
+          cache: maven
       - name: Check License Header
         uses: apache/skywalking-eyes/header@main
       - name: Check Dependencies' License
         uses: apache/skywalking-eyes/dependency@main
-  build:
+        env:
+          MAVEN_OPTS: -Xmx1g
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+  test:
     name: "BifroMQ Code Coverage Test"
-    needs: check-license
+    needs: license
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

.github/workflows/build-dev.yaml

@@ -14,18 +14,43 @@ on:
       - 'bugfix-**'
       - 'fix-**'
 jobs:
-  check-license:
-    name: "Check License"
+  prepare:
+    name: "Prepare"
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+      - name: Set up JDK 17
+        uses: actions/setup-java@v4
+        with:
+          java-version: '17'
+          distribution: 'temurin'
+          cache: maven
+      - name: Install without tests
+        run: mvn -q -DskipTests install
+
+  license:
+    name: "License Check"
+    needs: prepare
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up JDK 17
+        uses: actions/setup-java@v4
+        with:
+          java-version: '17'
+          distribution: 'temurin'
+          cache: maven
       - name: Check License Header
         uses: apache/skywalking-eyes/header@main
       - name: Check Dependencies' License
         uses: apache/skywalking-eyes/dependency@main
-  build:
+        env:
+          MAVEN_OPTS: -Xmx1g
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+  test:
     name: "Build BifroMQ"
-    needs: check-license
+    needs: license
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

.licenserc.yaml

@@ -25,6 +25,9 @@ header:
     - '.gitignore'
     - '.gitattributes'
     - 'lombok.config'
+    - 'META-INF/**'
+    - '**/META-INF/**'
+    - '**/resources/META-INF/**'
     - '**/*.crt'
     - '**/*.pem'
     - '**/MANIFEST.MF'

base-cluster/src/main/java/org/apache/bifromq/basecluster/memberlist/HostMemberList.java

@@ -381,6 +381,11 @@ private void handleJoin(Join join) {
                 messenger.send(ClusterMessage.newBuilder()
                     .setJoin(Join.newBuilder().setMember(local).build())
                     .build(), getMemberAddress(joinMember.getEndpoint()), true);
+            } else if (newMember) {
+                // send back a join to speed up convergence
+                messenger.send(ClusterMessage.newBuilder()
+                    .setJoin(Join.newBuilder().setMember(local).build())
+                    .build(), getMemberAddress(joinMember.getEndpoint()), true);
             }
         } else {
             clearZombie(join.getExpectedHost());

base-cluster/src/main/java/org/apache/bifromq/basecluster/memberlist/agent/CRDTUtil.java

@@ -14,7 +14,7 @@
  * "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.    
+ * under the License.
  */
 
 package org.apache.bifromq.basecluster.memberlist.agent;
@@ -23,10 +23,6 @@
 import static org.apache.bifromq.basecrdt.core.api.CausalCRDTType.mvreg;
 import static org.apache.bifromq.basecrdt.core.api.CausalCRDTType.ormap;
 
-import org.apache.bifromq.basecluster.agent.proto.AgentMemberAddr;
-import org.apache.bifromq.basecluster.agent.proto.AgentMemberMetadata;
-import org.apache.bifromq.basecrdt.core.api.IMVReg;
-import org.apache.bifromq.basecrdt.core.api.IORMap;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.google.protobuf.InvalidProtocolBufferException;
@@ -41,6 +37,10 @@
 import lombok.NoArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bifromq.basecluster.agent.proto.AgentMemberAddr;
+import org.apache.bifromq.basecluster.agent.proto.AgentMemberMetadata;
+import org.apache.bifromq.basecrdt.core.api.IMVReg;
+import org.apache.bifromq.basecrdt.core.api.IORMap;
 
 @Slf4j
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
@@ -56,7 +56,8 @@ static Map<AgentMemberAddr, AgentMemberMetadata> toAgentMemberMap(IORMap agentCR
         Iterator<IORMap.ORMapKey> orMapKeyItr = agentCRDT.keys();
         while (orMapKeyItr.hasNext()) {
             IORMap.ORMapKey orMapKey = orMapKeyItr.next();
-            agentMemberMap.put(parseAgentMemberAddr(orMapKey), parseMetadata(agentCRDT.getMVReg(orMapKey.key())).get());
+            Optional<AgentMemberMetadata> meta = parseMetadata(agentCRDT.getMVReg(orMapKey.key()));
+            meta.ifPresent(m -> agentMemberMap.put(parseAgentMemberAddr(orMapKey), m));
         }
         return agentMemberMap;
     }
@@ -78,11 +79,11 @@ private static Optional<AgentMemberMetadata> parseMetadata(IMVReg value) {
                     return AgentMemberMetadata.parseFrom(data);
                 } catch (InvalidProtocolBufferException e) {
                     log.error("Unable to parse agent host node", e);
-                    // this exception should not happen
+                    // should not happen, skip the broken value
                     return null;
                 }
             }), Objects::nonNull));
-        metaList.sort(Comparator.comparingLong(AgentMemberMetadata::getHlc));
+        metaList.sort(Comparator.comparingLong(AgentMemberMetadata::getHlc).reversed());
         return Optional.ofNullable(metaList.isEmpty() ? null : metaList.get(0));
     }
 }

base-cluster/src/main/java/org/apache/bifromq/basecluster/messenger/Messenger.java

@@ -14,17 +14,11 @@
  * "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.    
+ * under the License.
  */
 
 package org.apache.bifromq.basecluster.messenger;
 
-import org.apache.bifromq.basecluster.messenger.proto.DirectMessage;
-import org.apache.bifromq.basecluster.messenger.proto.GossipMessage;
-import org.apache.bifromq.basecluster.messenger.proto.MessengerMessage;
-import org.apache.bifromq.basecluster.proto.ClusterMessage;
-import org.apache.bifromq.basecluster.transport.ITransport;
-import org.apache.bifromq.basecluster.util.RandomUtils;
 import com.google.common.collect.Maps;
 import com.google.protobuf.InvalidProtocolBufferException;
 import io.micrometer.core.instrument.Counter;
@@ -46,26 +40,27 @@
 import java.util.stream.Collectors;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bifromq.basecluster.messenger.proto.DirectMessage;
+import org.apache.bifromq.basecluster.messenger.proto.GossipMessage;
+import org.apache.bifromq.basecluster.messenger.proto.MessengerMessage;
+import org.apache.bifromq.basecluster.proto.ClusterMessage;
+import org.apache.bifromq.basecluster.transport.ITransport;
+import org.apache.bifromq.basecluster.util.RandomUtils;
+import org.apache.bifromq.baseenv.ZeroCopyParser;
 
 @Slf4j
 public class Messenger implements IMessenger {
-    private enum State {
-        INIT, START, STOP
-    }
-
-    private State state = State.INIT;
     // threshold for determine which transport to use
     private final MessengerTransport transport;
     private final CompositeDisposable disposables = new CompositeDisposable();
     private final InetSocketAddress localAddress;
     private final Gossiper gossiper;
     private final Subject<Timed<MessageEnvelope>> publisher =
         PublishSubject.<Timed<MessageEnvelope>>create().toSerialized();
-
     private final Scheduler scheduler;
     private final MessengerOptions opts;
     private final MetricManager metricManager;
-
+    private State state = State.INIT;
     @Builder
     private Messenger(ITransport transport, Scheduler scheduler, MessengerOptions opts) {
         this.transport = new MessengerTransport(transport);
@@ -202,8 +197,8 @@ private void onMessengerMessage(Timed<MessengerMessageEnvelope> timedMessageEnve
         switch (messengerMessageEnvelope.message.getMessengerMessageTypeCase()) {
             case DIRECT:
                 try {
-                    ClusterMessage clusterMessage =
-                        ClusterMessage.parseFrom(messengerMessageEnvelope.message.getDirect().getPayload());
+                    ClusterMessage clusterMessage = ZeroCopyParser.parse(
+                        messengerMessageEnvelope.message.getDirect().getPayload(), ClusterMessage.parser());
                     log.trace("Received message: sender={}, message={}",
                         messengerMessageEnvelope.sender, clusterMessage);
                     metricManager.msgRecvCounters.get(clusterMessage.getClusterMessageTypeCase()).increment();
@@ -244,6 +239,10 @@ private void onError(Throwable throwable) {
         log.error("Received unexpected error:", throwable);
     }
 
+    private enum State {
+        INIT, START, STOP
+    }
+
     private static class MetricManager {
         final Map<ClusterMessage.ClusterMessageTypeCase, Counter> msgSendCounters = Maps.newHashMap();
         final Map<ClusterMessage.ClusterMessageTypeCase, Counter> msgRecvCounters = Maps.newHashMap();
@@ -253,17 +252,19 @@ private static class MetricManager {
 
         MetricManager(InetSocketAddress localAddress) {
             for (ClusterMessage.ClusterMessageTypeCase typeCase : ClusterMessage.ClusterMessageTypeCase.values()) {
-                Tags tags = Tags
-                    .of("local", localAddress.getAddress().getHostAddress() + ":" + localAddress.getPort())
-                    .and("type", typeCase.name());
-                msgSendCounters.put(typeCase,
-                    Metrics.counter("basecluster.send.count", tags));
-                msgRecvCounters.put(typeCase,
-                    Metrics.counter("basecluster.recv.count", tags));
-                gossipGenCounters.put(typeCase,
-                    Metrics.counter("basecluster.gossip.gen.count", tags));
-                gossipHeardCounters.put(typeCase,
-                    Metrics.counter("basecluster.gossip.heard.count", tags));
+                if (typeCase != ClusterMessage.ClusterMessageTypeCase.CLUSTERMESSAGETYPE_NOT_SET) {
+                    Tags tags = Tags
+                        .of("local", localAddress.getAddress().getHostAddress() + ":" + localAddress.getPort())
+                        .and("type", typeCase.name());
+                    msgSendCounters.put(typeCase,
+                        Metrics.counter("basecluster.send.count", tags));
+                    msgRecvCounters.put(typeCase,
+                        Metrics.counter("basecluster.recv.count", tags));
+                    gossipGenCounters.put(typeCase,
+                        Metrics.counter("basecluster.gossip.gen.count", tags));
+                    gossipHeardCounters.put(typeCase,
+                        Metrics.counter("basecluster.gossip.heard.count", tags));
+                }
             }
         }
 

base-cluster/src/main/java/org/apache/bifromq/basecluster/messenger/MessengerTransport.java

@@ -33,6 +33,7 @@
 import org.apache.bifromq.basecluster.messenger.proto.MessengerMessage;
 import org.apache.bifromq.basecluster.transport.ITransport;
 import org.apache.bifromq.basecluster.transport.PacketEnvelope;
+import org.apache.bifromq.baseenv.ZeroCopyParser;
 
 @Slf4j
 final class MessengerTransport {
@@ -62,9 +63,11 @@ Observable<Timed<MessengerMessageEnvelope>> receive() {
     private Observable<Timed<MessengerMessageEnvelope>> convert(PacketEnvelope packetEnvelope) {
         return Observable.fromIterable(packetEnvelope.data.stream().map(b -> {
             try {
+                // Parse with aliasing directly from ByteString to reduce copies
+                MessengerMessage msg = ZeroCopyParser.parse(b, MessengerMessage.parser());
                 MessengerMessageEnvelope mmEnvelop = MessengerMessageEnvelope.builder()
                     .recipient(packetEnvelope.recipient)
-                    .message(MessengerMessage.parseFrom(b))
+                    .message(msg)
                     .sender(packetEnvelope.sender)
                     .build();
                 return new Timed<>(mmEnvelop, System.currentTimeMillis(), TimeUnit.MILLISECONDS);

base-cluster/src/main/java/org/apache/bifromq/basecluster/transport/UDPTransport.java

@@ -14,14 +14,11 @@
  * "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.    
+ * under the License.
  */
 
 package org.apache.bifromq.basecluster.transport;
 
-import org.apache.bifromq.basecluster.transport.proto.Packet;
-import org.apache.bifromq.baseenv.NettyEnv;
-import org.apache.bifromq.basehlc.HLC;
 import com.google.protobuf.InvalidProtocolBufferException;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
@@ -49,6 +46,10 @@
 import lombok.Builder;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bifromq.basecluster.transport.proto.Packet;
+import org.apache.bifromq.baseenv.NettyEnv;
+import org.apache.bifromq.baseenv.ZeroCopyParser;
+import org.apache.bifromq.basehlc.HLC;
 
 @Slf4j
 public final class UDPTransport extends AbstractTransport {
@@ -151,7 +152,8 @@ protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket dp) {
             try {
                 byte[] data = new byte[dp.content().readableBytes()];
                 dp.content().readBytes(data);
-                Packet packet = Packet.parseFrom(data);
+                // Parse with aliasing to avoid extra copies of bytes fields
+                Packet packet = ZeroCopyParser.parse(data, Packet.parser());
                 transportLatency.record(HLC.INST.getPhysical(packet.getHlc() - HLC.INST.get()));
                 doReceive(packet, dp.sender(), dp.recipient());
             } catch (InvalidProtocolBufferException e) {