log: add details for onError

Author: ffffwh

driver/kafka/kafka3.go

@@ -634,6 +634,8 @@ func (kr *KafkaRunner) initiateStreaming() error {
 }
 
 func (kr *KafkaRunner) onError(state int, err error) {
+	kr.logger.Info("onError", "err", err, "hasShutdown", kr.shutdown)
+
 	if kr.shutdown {
 		return
 	}
@@ -666,7 +668,7 @@ func (kr *KafkaRunner) onError(state int, err error) {
 		OOMKilled: false,
 		Err:       err,
 	}
-	kr.Shutdown()
+	_ = kr.Shutdown()
 }
 
 func (kr *KafkaRunner) kafkaTransformSnapshotData(

driver/mysql/applier.go

@@ -1019,7 +1019,7 @@ func (a *Applier) Stats() (*common.TaskStatistics, error) {
 }
 
 func (a *Applier) onError(state int, err error) {
-	a.logger.Error("onError", "err", err)
+	a.logger.Error("onError", "err", err, "hasShutdown", a.shutdown)
 	if a.shutdown {
 		return
 	}

driver/mysql/extractor.go

@@ -1560,7 +1560,7 @@ func (e *Extractor) Stats() (*common.TaskStatistics, error) {
 }
 
 func (e *Extractor) onError(state int, err error) {
-	e.logger.Error("onError", "err", err)
+	e.logger.Error("onError", "err", err, "hasShutdown", e.shutdown)
 	if e.shutdown {
 		return
 	}

driver/oracle/extractor/extractor_oracle.go

@@ -817,7 +817,7 @@ func (e *ExtractorOracle) publish(subject string, txMsg []byte, gno int64) (err
 }
 
 func (e *ExtractorOracle) onError(state int, err error) {
-	e.logger.Error("onError", "err", err)
+	e.logger.Error("onError", "err", err, "hasShutdown", e.shutdown)
 	if e.shutdown {
 		return
 	}