feat(mysql): 优化 mysql 慢查询视图 #10235

Author: seanlook

dbm-services/common/bkdata-kafka-consumer/pkg/consumer/mysql_table_size.go

@@ -92,10 +92,10 @@ func (c *MysqlTableSize) ConsumeClaim(session sarama.ConsumerGroupSession, claim
 				if err := c.HandleMessageTryBatch(items, c.Sinker, c.Db); err != nil {
 					slog.Error("handle message batch", err)
 				} else {
+					slog.Info("sink message", slog.String("key", string(msgs[len(msgs)-1].Key)))
 					session.MarkMessage(msgs[len(msgs)-1], "")
 					items = items[:0]
 					msgs = msgs[:0]
-					slog.Info("sink message", slog.String("key", string(msgs[len(msgs)-1].Key)))
 				}
 			}
 		case message := <-claim.Messages():
@@ -127,10 +127,10 @@ func (c *MysqlTableSize) ConsumeClaim(session sarama.ConsumerGroupSession, claim
 					slog.Error("handle message batch", err)
 					time.Sleep(200 * time.Millisecond)
 				} else {
+					slog.Info("sink message", slog.String("key", string(message.Key)))
 					session.MarkMessage(message, "")
 					items = items[:0]
 					msgs = msgs[:0]
-					slog.Info("sink message", slog.String("key", string(message.Key)))
 				}
 			}
 		case <-session.Context().Done():

dbm-services/common/bkdata-kafka-consumer/pkg/consumer/mysql_table_size_for_doris.go

@@ -69,10 +69,10 @@ func (c *MysqlTableSizeDoris) ConsumeClaim(session sarama.ConsumerGroupSession,
 				if err := c.HandleMessageTryBatch(items, c.Sinker, c.Db); err != nil {
 					slog.Error("handle message batch", err)
 				} else {
+					slog.Info("sink message", slog.String("key", string(msgs[len(msgs)-1].Key)))
 					session.MarkMessage(msgs[len(msgs)-1], "")
 					items = items[:0]
 					msgs = msgs[:0]
-					slog.Info("sink message", slog.String("key", string(msgs[len(msgs)-1].Key)))
 				}
 			}
 		case message := <-claim.Messages():
@@ -104,10 +104,10 @@ func (c *MysqlTableSizeDoris) ConsumeClaim(session sarama.ConsumerGroupSession,
 					slog.Error("handle message batch", err)
 					time.Sleep(200 * time.Millisecond)
 				} else {
+					slog.Info("sink message", slog.String("key", string(message.Key)))
 					session.MarkMessage(message, "")
 					items = items[:0]
 					msgs = msgs[:0]
-					slog.Info("sink message", slog.String("key", string(message.Key)))
 				}
 			}
 		case <-session.Context().Done():