msg update

internal/msgtransfer/online_history_msg_handler.go

@@ -205,7 +205,8 @@ func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerG
 		}
 	}
 	rwLock := new(sync.RWMutex)
-	log.ZDebug(context.Background(), "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
+	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
+		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
 	cMsg := make([]*sarama.ConsumerMessage, 0, 1000)
 	t := time.NewTicker(time.Duration(100) * time.Millisecond)
 	go func() {

internal/push/push_handler.go

@@ -28,7 +28,6 @@ func NewConsumerHandler(pusher *Pusher) *ConsumerHandler {
 }
 
 func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
-	log.ZDebug(ctx, "msg come from kafka  And push!!!", "msg", string(msg))
 	msgFromMQ := pbChat.PushMsgDataToMQ{}
 	if err := proto.Unmarshal(msg, &msgFromMQ); err != nil {
 		log.ZError(ctx, "push Unmarshal msg err", err, "msg", string(msg))
@@ -59,7 +58,6 @@ func (ConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil
 func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
 	claim sarama.ConsumerGroupClaim) error {
 	for msg := range claim.Messages() {
-		log.NewDebug("", "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
 		ctx := c.pushConsumerGroup.GetContextFromMsg(msg, "push consumer")
 		c.handleMs2PsChat(ctx, msg.Value)
 		sess.MarkMessage(msg, "")