morpc: active checksum in cn, dn and logservice (#4592)

pkg/logservice/client.go

@@ -517,7 +517,7 @@ func getRPCClient(ctx context.Context, target string, pool *sync.Pool) (morpc.RP
 	// we set connection timeout to a constant value so if ctx's deadline is much
 	// larger, then we can ensure that all specified potential nodes have a chance
 	// to be attempted
-	codec := morpc.NewMessageCodec(mf, defaultWriteSocketSize)
+	codec := morpc.NewMessageCodecWithChecksum(mf, defaultWriteSocketSize)
 	bf := morpc.NewGoettyBasedBackendFactory(codec, backendOpts...)
 	return morpc.NewClient(bf, clientOpts...)
 }

pkg/logservice/service.go

@@ -99,7 +99,7 @@ func NewService(cfg Config, opts ...Option) (*Service, error) {
 		return pool.Get().(*RPCRequest)
 	}
 	// TODO: check and fix all these magic numbers
-	codec := morpc.NewMessageCodec(mf, 16*1024)
+	codec := morpc.NewMessageCodecWithChecksum(mf, 16*1024)
 	server, err := morpc.NewRPCServer(LogServiceRPCName, cfg.ServiceListenAddress, codec,
 		morpc.WithServerGoettyOptions(goetty.WithSessionReleaseMsgFunc(func(i interface{}) {
 			respPool.Put(i.(morpc.RPCMessage).Message)

pkg/txn/rpc/sender.go

@@ -102,7 +102,7 @@ func NewSender(logger *zap.Logger, options ...SenderOption) (TxnSender, error) {
 		},
 	}
 
-	codec := morpc.NewMessageCodec(func() morpc.Message { return s.acquireResponse() },
+	codec := morpc.NewMessageCodecWithChecksum(func() morpc.Message { return s.acquireResponse() },
 		s.options.payloadCopyBufferSize)
 	bf := morpc.NewGoettyBasedBackendFactory(codec, s.options.backendCreateOptions...)
 	client, err := morpc.NewClient(bf, s.options.clientOptions...)

pkg/txn/rpc/sender_test.go

@@ -256,7 +256,7 @@ func TestNewSenderWithOptions(t *testing.T) {
 
 func newTestTxnServer(t assert.TestingT, addr string) morpc.RPCServer {
 	assert.NoError(t, os.RemoveAll(addr[7:]))
-	codec := morpc.NewMessageCodec(func() morpc.Message { return &txn.TxnRequest{} }, 0)
+	codec := morpc.NewMessageCodecWithChecksum(func() morpc.Message { return &txn.TxnRequest{} }, 0)
 	s, err := morpc.NewRPCServer("test-txn-server", addr, codec)
 	assert.NoError(t, err)
 	assert.NoError(t, s.Start())

pkg/txn/rpc/server.go

@@ -58,7 +58,7 @@ func NewTxnServer(address string, logger *zap.Logger) (TxnServer, error) {
 	}
 
 	rpc, err := morpc.NewRPCServer("txn-server", address,
-		morpc.NewMessageCodec(s.acquireRequest, 16*1024),
+		morpc.NewMessageCodecWithChecksum(s.acquireRequest, 16*1024),
 		morpc.WithServerLogger(s.logger),
 		morpc.WithServerGoettyOptions(goetty.WithSessionReleaseMsgFunc(func(v interface{}) {
 			m := v.(morpc.RPCMessage)