mq: migrate broker consumers off gRPC to ZAP

- broker SubscribeMessage: dial the follower broker over transport.Dial and open
  the SubscribeFollowMe client-stream via OpenStream(...Ordinal), forwarding Ack/
  Close as zero-copy wire frames (mirrors PublishFollowMe in local_partition.go);
  drops pb.GrpcDial + mq_pb.NewHanzoMessagingClient.
- kafka integration BrokerClient: replace grpc.DialContext + NewHanzoMessagingClient
  with dialBrokerZap (PQ-TLS/plaintext) + mqzap.New; conn is now transport.Conn.

s3/mq/broker/broker_grpc_sub.go

@@ -6,14 +6,17 @@ import (
 	"io"
 	"time"
 
+	"github.com/zap-proto/go/transport"
+
 	"github.com/hanzoai/s3/s3/glog"
+	"github.com/hanzoai/s3/s3/mq/agent/agentconv"
 	"github.com/hanzoai/s3/s3/mq/sub_coordinator"
 	"github.com/hanzoai/s3/s3/mq/topic"
-	"github.com/hanzoai/s3/s3/pb"
 	"github.com/hanzoai/s3/s3/pb/filer_pb"
 	"github.com/hanzoai/s3/s3/pb/mq_pb"
 	"github.com/hanzoai/s3/s3/pb/schema_pb"
 	"github.com/hanzoai/s3/s3/util/log_buffer"
+	mq_brokerwire "github.com/hanzoai/s3/s3/wire/mq_broker"
 )
 
 func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.HanzoMessaging_SubscribeMessageServer) error {
@@ -72,37 +75,36 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.HanzoMessaging_Subscr
 		}
 	}()
 
-	// connect to the follower
-	var subscribeFollowMeStream mq_pb.HanzoMessaging_SubscribeFollowMeClient
+	// connect to the follower over the native ZAP transport. The follower broker
+	// fully cut over to ZAP (no gRPC), so its address IS the ZAP endpoint: dial it,
+	// open the SubscribeFollowMe client-stream with the Init frame, then forward Ack
+	// and Close as zero-copy wire frames — same doctrine as the PublishFollowMe
+	// follower path in local_partition.go.
+	var subscribeFollowMeStream transport.Stream
+	var followerConn transport.Conn
 	glog.V(0).Infof("follower broker: %v", req.GetInit().FollowerBroker)
 	if req.GetInit().FollowerBroker != "" {
 		follower := req.GetInit().FollowerBroker
-		if followerGrpcConnection, err := pb.GrpcDial(ctx, follower, true, b.grpcDialOption); err != nil {
+		if followerConn, err = transport.Dial("tcp", follower); err != nil {
 			return fmt.Errorf("fail to dial %s: %v", follower, err)
-		} else {
-			defer func() {
-				println("closing SubscribeFollowMe connection", follower)
-				if subscribeFollowMeStream != nil {
-					subscribeFollowMeStream.CloseSend()
-				}
-				// followerGrpcConnection.Close()
-			}()
-			followerClient := mq_pb.NewHanzoMessagingClient(followerGrpcConnection)
-			if subscribeFollowMeStream, err = followerClient.SubscribeFollowMe(ctx); err != nil {
-				return fmt.Errorf("fail to subscribe to %s: %v", follower, err)
-			} else {
-				if err := subscribeFollowMeStream.Send(&mq_pb.SubscribeFollowMeRequest{
-					Message: &mq_pb.SubscribeFollowMeRequest_Init{
-						Init: &mq_pb.SubscribeFollowMeRequest_InitMessage{
-							Topic:         req.GetInit().Topic,
-							Partition:     req.GetInit().GetPartitionOffset().Partition,
-							ConsumerGroup: req.GetInit().ConsumerGroup,
-						},
-					},
-				}); err != nil {
-					return fmt.Errorf("fail to send init to %s: %v", follower, err)
-				}
+		}
+		defer func() {
+			println("closing SubscribeFollowMe connection", follower)
+			if subscribeFollowMeStream != nil {
+				_ = subscribeFollowMeStream.CloseSend()
 			}
+			_ = followerConn.Close()
+		}()
+		initBuf := mq_brokerwire.NewSubscribeFollowMeRequest(mq_brokerwire.SubscribeFollowMeRequestInput{
+			MessageWhich: mq_brokerwire.SubscribeFollowMeRequestMessageInit,
+			MessageValue: mq_brokerwire.NewSubscribeFollowMeRequestInitMessage(mq_brokerwire.SubscribeFollowMeRequestInitMessageInput{
+				Topic:         agentconv.TopicToWire(req.GetInit().Topic),
+				Partition:     agentconv.PartitionToWire(req.GetInit().GetPartitionOffset().Partition),
+				ConsumerGroup: req.GetInit().ConsumerGroup,
+			}),
+		})
+		if subscribeFollowMeStream, err = followerConn.OpenStream(mq_brokerwire.HanzoMessagingSubscribeFollowMeOrdinal, initBuf); err != nil {
+			return fmt.Errorf("fail to subscribe to %s: %v", follower, err)
 		}
 		glog.V(0).Infof("follower %s connected", follower)
 	}
@@ -163,13 +165,12 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.HanzoMessaging_Subscr
 			subscriber.UpdateAckedOffset(currentLastOffset)
 			// fmt.Printf("%+v recv (%s,%d), oldest %d\n", partition, string(ack.GetAck().Key), ack.GetAck().TsNs, currentLastOffset)
 			if subscribeFollowMeStream != nil && currentLastOffset > lastOffset {
-				if err := subscribeFollowMeStream.Send(&mq_pb.SubscribeFollowMeRequest{
-					Message: &mq_pb.SubscribeFollowMeRequest_Ack{
-						Ack: &mq_pb.SubscribeFollowMeRequest_AckMessage{
-							TsNs: currentLastOffset,
-						},
-					},
-				}); err != nil {
+				if err := subscribeFollowMeStream.Send(mq_brokerwire.NewSubscribeFollowMeRequest(mq_brokerwire.SubscribeFollowMeRequestInput{
+					MessageWhich: mq_brokerwire.SubscribeFollowMeRequestMessageAck,
+					MessageValue: mq_brokerwire.NewSubscribeFollowMeRequestAckMessage(mq_brokerwire.SubscribeFollowMeRequestAckMessageInput{
+						TsNs: currentLastOffset,
+					}),
+				})); err != nil {
 					glog.Errorf("Error sending ack to follower: %v", err)
 					break
 				}
@@ -184,11 +185,10 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.HanzoMessaging_Subscr
 			}
 		}
 		if subscribeFollowMeStream != nil {
-			if err := subscribeFollowMeStream.Send(&mq_pb.SubscribeFollowMeRequest{
-				Message: &mq_pb.SubscribeFollowMeRequest_Close{
-					Close: &mq_pb.SubscribeFollowMeRequest_CloseMessage{},
-				},
-			}); err != nil {
+			if err := subscribeFollowMeStream.Send(mq_brokerwire.NewSubscribeFollowMeRequest(mq_brokerwire.SubscribeFollowMeRequestInput{
+				MessageWhich: mq_brokerwire.SubscribeFollowMeRequestMessageClose,
+				MessageValue: mq_brokerwire.NewSubscribeFollowMeRequestCloseMessage(mq_brokerwire.SubscribeFollowMeRequestCloseMessageInput{}),
+			})); err != nil {
 				if err != io.EOF {
 					glog.Errorf("Error sending close to follower: %v", err)
 				}

s3/mq/kafka/integration/broker_client.go

@@ -10,9 +10,12 @@ import (
 
 	"google.golang.org/grpc"
 
+	"github.com/zap-proto/go/transport"
+
 	"github.com/hanzoai/s3/s3/filer_client"
 	"github.com/hanzoai/s3/s3/glog"
 	"github.com/hanzoai/s3/s3/mq"
+	"github.com/hanzoai/s3/s3/mqzap"
 	"github.com/hanzoai/s3/s3/pb"
 	"github.com/hanzoai/s3/s3/pb/filer_pb"
 	"github.com/hanzoai/s3/s3/pb/mq_pb"
@@ -27,35 +30,34 @@ const (
 	brokerHealthCheckBackoff  = 250 * time.Millisecond
 )
 
+// dialBrokerZap opens a ZAP connection to a broker address, PQ-secured mTLS when
+// grpc.mq.cert/.key is configured and plaintext otherwise — the single place this
+// package's broker dial-side security gate lives, mirroring pb.dialBrokerZapAddr.
+func dialBrokerZap(brokerAddress string) (transport.Conn, error) {
+	if cfg := security.ClientTLSConfig(util.GetViper(), "grpc.mq"); cfg != nil {
+		return transport.DialTLS("tcp", brokerAddress, transport.PQTLSConfig(cfg))
+	}
+	return transport.Dial("tcp", brokerAddress)
+}
+
 // NewBrokerClientWithFilerAccessor creates a client with a shared filer accessor
 func NewBrokerClientWithFilerAccessor(brokerAddress string, filerClientAccessor *filer_client.FilerClientAccessor) (*BrokerClient, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	// Use background context for gRPC connections to prevent them from being canceled
-	// when BrokerClient.Close() is called. This allows subscriber streams to continue
-	// operating even during client shutdown, which is important for testing scenarios.
-	dialCtx := context.Background()
-
-	// CRITICAL FIX: Add timeout to dial context
-	// gRPC dial will retry with exponential backoff. Without a timeout, it hangs indefinitely
-	// if the broker is unreachable. Set a reasonable timeout for initial connection attempt.
-	dialCtx, dialCancel := context.WithTimeout(dialCtx, 30*time.Second)
-	defer dialCancel()
-
-	// Connect to broker
-	// Load security configuration for broker connection
+	// Connect to the broker over the native ZAP transport. The broker fully cut over
+	// to ZAP (no gRPC), so its address IS the ZAP endpoint. When grpc.mq.cert/.key is
+	// configured the connection is PQ-secured mTLS (transport.PQTLSConfig pins the
+	// X25519MLKEM768 hybrid); otherwise plaintext (loopback / dev), matching the
+	// broker server's gating. The Conn outlives BrokerClient.Close of individual
+	// streams: closing it (in Close) drops every multiplexed stream at once.
 	util.LoadSecurityConfiguration()
-	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.mq")
-
-	conn, err := grpc.DialContext(dialCtx, brokerAddress,
-		grpcDialOption,
-	)
+	conn, err := dialBrokerZap(brokerAddress)
 	if err != nil {
 		cancel()
 		return nil, fmt.Errorf("failed to connect to broker %s: %v", brokerAddress, err)
 	}
 
-	client := mq_pb.NewHanzoMessagingClient(conn)
+	client := mqzap.New(conn, nil)
 
 	return &BrokerClient{
 		filerClientAccessor:         filerClientAccessor,

s3/mq/kafka/integration/types.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 	"time"
 
-	"google.golang.org/grpc"
+	"github.com/zap-proto/go/transport"
 
 	"github.com/hanzoai/s3/s3/filer_client"
 	"github.com/hanzoai/s3/s3/pb/mq_pb"
@@ -172,7 +172,7 @@ type BrokerClient struct {
 	filerClientAccessor *filer_client.FilerClientAccessor
 
 	brokerAddress string
-	conn          *grpc.ClientConn
+	conn          transport.Conn
 	client        mq_pb.HanzoMessagingClient
 
 	// Publisher streams: topic-partition -> stream info