add logs

- info 'applier tx committed' if env set
  - docker run -e DTLE_LOG_TX_COMMIT=1 actiontech/dtle
- detail: unexpected error on publish

Author: ffffwh

driver/mysql/applier_incr.go

@@ -71,7 +71,8 @@ type ApplierIncr struct {
 	sourceType  string
 	tableSpecs  []*common.TableSpec
 
-	inBigTx bool
+	inBigTx     bool
+	logTxCommit bool
 }
 
 func NewApplierIncr(applier *Applier, sourcetype string) (*ApplierIncr, error) {
@@ -99,6 +100,10 @@ func NewApplierIncr(applier *Applier, sourcetype string) (*ApplierIncr, error) {
 		a.SkipGtidExecutedTable = true
 	}
 
+	if g.EnvIsTrue(g.ENV_DTLE_LOG_TX_COMMIT) {
+		a.logTxCommit = true
+	}
+
 	a.timestampCtx = NewTimestampContext(a.shutdownCh, a.logger, func() bool {
 		return len(a.binlogEntryQueue) == 0 && len(a.applyBinlogMtsTxQueue) == 0
 		// TODO need a more reliable method to determine queue.empty.
@@ -732,7 +737,11 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 		if a.printTps {
 			atomic.AddUint32(&a.txLastNSeconds, 1)
 		}
-		logger.Debug("applier tx committed", "gno", binlogEntry.Coordinates.GetGNO())
+		if a.logTxCommit {
+			logger.Info("applier tx committed", "gno", binlogEntry.Coordinates.GetGNO())
+		} else {
+			logger.Debug("applier tx committed", "gno", binlogEntry.Coordinates.GetGNO())
+		}
 		atomic.AddUint32(&a.appliedTxCount, 1)
 	}
 	a.EntryExecutedHook(binlogEntry)

driver/mysql/extractor.go

@@ -1104,7 +1104,8 @@ func (e *Extractor) publish(subject string, txMsg []byte, gno int64) (err error)
 		e.logger.Debug("publish", "subject", subject, "gno", gno, "partLen", len(part), "iSeg", iSeg)
 		_, err := e.natsConn.Request(subject, part, 24*time.Hour)
 		if err != nil {
-			e.logger.Error("unexpected error on publish", "err", err)
+			e.logger.Error("unexpected error on publish", "err", err,
+				"subject", subject, "gno", gno, "partLen", len(part), "iSeg", iSeg)
 			return err
 		}
 	}

g/g.go

@@ -35,15 +35,16 @@ const (
 
 	JobNameLenLimit = 64
 
-	ENV_PRINT_TPS         = "UDUP_PRINT_TPS"
-	ENV_DUMP_CHECKSUM     = "DTLE_DUMP_CHECKSUM"
-	ENV_DUMP_OLDWAY       = "DTLE_DUMP_OLDWAY"
-	ENV_TESTSTUB1_DELAY   = "UDUP_TESTSTUB1_DELAY"
-	ENV_FULL_APPLY_DELAY  = "DTLE_FULL_APPLY_DELAY"
-	ENV_COUNT_INFO_SCHEMA = "DTLE_COUNT_INFO_SCHEMA"
-	ENV_BIG_MSG_100K      = "DTLE_BIG_MSG_100K"
+	ENV_PRINT_TPS                = "UDUP_PRINT_TPS"
+	ENV_DUMP_CHECKSUM            = "DTLE_DUMP_CHECKSUM"
+	ENV_DUMP_OLDWAY              = "DTLE_DUMP_OLDWAY"
+	ENV_TESTSTUB1_DELAY          = "UDUP_TESTSTUB1_DELAY"
+	ENV_FULL_APPLY_DELAY         = "DTLE_FULL_APPLY_DELAY"
+	ENV_BIG_MSG_100K             = "DTLE_BIG_MSG_100K"
 	ENV_SKIP_GTID_EXECUTED_TABLE = "DTLE_SKIP_GTID_EXECUTED_TABLE"
-	ENV_FORCE_MTS         = "DTLE_FORCE_MTS"
+	ENV_FORCE_MTS                = "DTLE_FORCE_MTS"
+	ENV_DTLE_LOG_TX_COMMIT       = "DTLE_LOG_TX_COMMIT"
+
 	NatsMaxPayload        = 64 * 1024 * 1024
 
 	LONG_LOG_LIMIT = 256