Repository: lf-edge/ekuiper
Pull request #1379 (2022-08-22 03:36:09)

Connection exception management


Patch:
diff --git a/internal/topo/connection/clients/mqtt/mqtt.go b/internal/topo/connection/clients/mqtt/mqtt.go
index 7ebfeb40..ae816460 100644
--- a/internal/topo/connection/clients/mqtt/mqtt.go
+++ b/internal/topo/connection/clients/mqtt/mqtt.go
@@ -100,6 +100,7 @@ func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
 func (ms *MQTTClient) Connect(connHandler MQTT.OnConnectHandler, lostHandler MQTT.ConnectionLostHandler) error {
 	if conf.Config.Basic.Debug {
 		MQTT.DEBUG = conf.Log
+		MQTT.ERROR = conf.Log
 	}
 	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(4)
 
@@ -115,6 +116,9 @@ func (ms *MQTTClient) Connect(connHandler MQTT.OnConnectHandler, lostHandler MQT
 	opts = opts.SetAutoReconnect(true)
 	opts.OnConnect = connHandler
 	opts.OnConnectionLost = lostHandler
+	opts.OnReconnecting = func(MQTT.Client, *MQTT.ClientOptions) {
+		conf.Log.Infof("Reconnecting to mqtt broker %s client id %s", ms.srv, ms.clientid)
+	}
 
 	c := MQTT.NewClient(opts)
 	if token := c.Connect(); token.Wait() && token.Error() != nil {
diff --git a/internal/topo/connection/clients/mqtt/mqtt_wrapper.go b/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
index 9386265f..499b6c60 100644
--- a/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
+++ b/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
@@ -103,7 +103,21 @@ func (mc *mqttClientWrapper) onConnectLost(_ pahoMqtt.Client, err error) {
 	mc.subLock.Lock()
 	defer mc.subLock.Unlock()
 	mc.connected = false
-	conf.Log.Warnf("The connection to mqtt broker %s client id %s disconnected with error: %s ", mc.cli.srv, mc.cli.clientid, err.Error())
+	e := fmt.Errorf("The connection to mqtt broker %s client id %s disconnected with error: %s ", mc.cli.srv, mc.cli.clientid, err.Error())
+	conf.Log.Warnf(e.Error())
+	for _, sub := range mc.topicSubscriptions {
+		if sub != nil {
+			// broadcast errors to all consumers
+			for _, consumer := range sub.topicConsumers {
+				select {
+				case consumer.SubErrors <- e:
+					break
+				default:
+					conf.Log.Warnf("consumer chan full for request id %s", consumer.ConsumerId)
+				}
+			}
+		}
+	}
 }
 
 func (mc *mqttClientWrapper) newMessageHandler(sub *mqttSubscriptionInfo) pahoMqtt.MessageHandler {
diff --git a/internal/topo/memory/manager.go b/internal/topo/memory/manager.go
index 0c26d333..28e4ac58 100644
--- a/internal/topo/memory/manager.go
+++ b/internal/topo/memory/manager.go
@@ -16,6 +16,7 @@ package memory
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"regexp"
 	"sync"
@@ -138,6 +139,28 @@ func Produce(ctx api.StreamContext, topic string, data map[string]interface{}) {
 
 }
 
+func ProduceError(ctx api.StreamContext, topic string, err error) {
+	c, exists := pubTopics[topic]
+	if !exists {
+		return
+	}
+	logger := ctx.GetLogger()
+	mu.RLock()
+	defer mu.RUnlock()
+	// broadcast to all consumers
+	for name, out := range c.consumers {
+		select {
+		case out <- &xsql.ErrorSourceTuple{Error: err}:
+			logger.Debugf("memory source broadcast error from topic %s to %s done", topic, name)
+		case <-ctx.Done():
+			// rule stop so stop waiting
+		default:
+			logger.Errorf("memory source topic %s drop message to %s", topic, name)
+		}
+	}
+
+}
+
 func addPubConsumer(topic string, sourceId string, ch chan api.SourceTuple) {
 	var sinkConsumerChannels *pubConsumers
 	if c, exists := pubTopics[topic]; exists {
diff --git a/internal/topo/mock/mock.go b/internal/topo/mock/mock.go
index c934c9be..e06565bb 100644
--- a/internal/topo/mock/mock.go
+++ b/internal/topo/mock/mock.go
@@ -21,7 +21,7 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
-func newMockContext(ruleId string, opId string) api.StreamContext {
+func NewMockContext(ruleId string, opId string) api.StreamContext {
 	contextLogger := conf.Log.WithField("rule", ruleId)
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	tempStore, _ := state.CreateStore(ruleId, api.AtMostOnce)
diff --git a/internal/topo/neuron/connection.go b/internal/topo/neuron/connection.go
index b31b72ea..61987c58 100644
--- a/internal/topo/neuron/connection.go
+++ b/internal/topo/neuron/connection.go
@@ -22,10 +22,12 @@ import (
 	"github.com/lf-edge/ekuiper/internal/topo/memory"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"go.nanomsg.org/mangos/v3"
 	"go.nanomsg.org/mangos/v3/protocol/pair"
 	_ "go.nanomsg.org/mangos/v3/transport/ipc"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -38,6 +40,7 @@ var (
 	m               sync.RWMutex
 	connectionCount int
 	sock            mangos.Socket
+	opened          int32
 	sendTimeout     = 100
 )
 
@@ -50,11 +53,6 @@ func createOrGetConnection(sc api.StreamContext, url string) error {
 	sc.GetLogger().Infof("createOrGetConnection count: %d", connectionCount)
 	if connectionCount == 0 {
 		sc.GetLogger().Infof("Creating neuron connection")
-		err := connect(url)
-		if err != nil {
-			return err
-		}
-		sc.GetLogger().Infof("Neuron connected")
 		contextLogger := conf.Log.WithField("neuron_connection", 0)
 		ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
 		ruleId := "$$neuron_connection"
@@ -65,6 +63,11 @@ func createOrGetConnection(sc api.StreamContext, url string) error {
 			return err
 		}
 		sctx := ctx.WithMeta(ruleId, opId, store)
+		err = connect(sctx, url)
+		if err != nil {
+			return err
+		}
+		sc.GetLogger().Infof("Neuron connected")
 		memory.CreatePub(NeuronTopic)
 		go run(sctx)
 	}
@@ -90,7 +93,7 @@ func closeConnection(ctx api.StreamContext, url string) error {
 // nng connections
 
 // connect to nng
-func connect(url string) error {
+func connect(ctx api.StreamContext, url string) error {
 	var err error
 	sock, err = pair.NewSocket()
 	if err != nil {
@@ -101,6 +104,19 @@ func connect(url string) error {
 	if err != nil {
 		return err
 	}
+	sock.SetPipeEventHook(func(ev mangos.PipeEvent, p mangos.Pipe) {
+		switch ev {
+		case mangos.PipeEventAttached:
+			atomic.StoreInt32(&opened, 1)
+			conf.Log.Infof("neuron connection attached")
+		case mangos.PipeEventAttaching:
+			conf.Log.Infof("neuron connection is attaching")
+		case mangos.PipeEventDetached:
+			atomic.StoreInt32(&opened, 0)
+			conf.Log.Warnf("neuron connection detached")
+			memory.ProduceError(ctx, NeuronTopic, fmt.Errorf("neuron connection detached"))
+		}
+	})
 	//sock.SetOption(mangos.OptionWriteQLen, 100)
 	//sock.SetOption(mangos.OptionReadQLen, 100)
 	//sock.SetOption(mangos.OptionBestEffort, false)
@@ -141,10 +157,10 @@ func run(ctx api.StreamContext) {
 
 func publish(ctx api.StreamContext, data []byte) error {
 	ctx.GetLogger().Debugf("publish to neuron: %s", string(data))
-	if sock != nil {
+	if sock != nil && atomic.LoadInt32(&opened) == 1 {
 		return sock.Send(data)
 	}
-	return fmt.Errorf("neuron connection is not established")
+	return fmt.Errorf("%s: neuron connection is not established", errorx.IOErr)
 }
 
 func disconnect(_ string) error {
diff --git a/internal/topo/neuron/sink.go b/internal/topo/neuron/sink.go
index 0b5c7b53..beed3bc8 100644
--- a/internal/topo/neuron/sink.go
+++ b/internal/topo/neuron/sink.go
@@ -78,22 +78,20 @@ func (s *sink) Collect(ctx api.StreamContext, data interface{}) error {
 			return err
 		}
 		return publish(ctx, r)
-
 	} else {
 		switch d := data.(type) {
 		case []map[string]interface{}:
 			for _, el := range d {
 				err := s.SendMapToNeuron(ctx, el)
 				if err != nil {
-					ctx.GetLogger().Errorf("Error sending map to neuron: %v", err)
+					return err
 				}
 			}
 			return nil
 		case map[string]interface{}:
 			return s.SendMapToNeuron(ctx, d)
 		default:
-			ctx.GetLogger().Errorf("unrecognized format of %s", data)
-			return nil
+			return fmt.Errorf("unrecognized format of %s", data)
 		}
 	}
 }
@@ -148,6 +146,7 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 			}
 		}
 	} else {
+		// Send as many tags as possible in order and drop the tag if it is invalid
 		for _, tag := range s.c.Tags {
 			t.TagName, err = ctx.ParseTemplate(tag, el)
 			if err != nil {
diff --git a/internal/topo/node/join_align_node.go b/internal/topo/node/join_align_node.go
index be94ad04..21fd937a 100644
--- a/internal/topo/node/join_align_node.go
+++ b/internal/topo/node/join_align_node.go
@@ -99,13 +99,13 @@ func (n *JoinAlignNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 					n.statManager.IncTotalRecordsIn()
 					n.statManager.ProcessTimeStart()
 					if !opened {
-						n.statManager.IncTotalExceptions()
+						n.statManager.IncTotalExceptions("input channel closed")
 						break
 					}
 					switch d := item.(type) {
 					case error:
 						n.Broadcast(d)
-						n.statManager.IncTotalExceptions()
+						n.statManager.IncTotalExceptions(d.Error())
 					case *xsql.Tuple:
 						log.Debugf("JoinAlignNode receive tuple input %s", d)
 						temp := &xsql.WindowTuples{
@@ -123,8 +123,9 @@ func (n *JoinAlignNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 							// Buffer and update batch inputs
 							_, ok := n.batch[emitter]
 							if !ok {
-								n.Broadcast(fmt.Errorf("run JoinAlignNode error: receive batch input from unknown emitter %[1]T(%[1]v)", d))
-								n.statManager.IncTotalExceptions()
+								e := fmt.Errorf("run JoinAlignNode error: receive batch input from unknown emitter %[1]T(%[1]v)", d)
+								n.Broadcast(e)
+								n.statManager.IncTotalExceptions(e.Error())
 								break
 							} else {
 								n.batch[emitter] = d.Content
@@ -132,8 +133,9 @@ func (n *JoinAlignNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 							}
 						}
 					default:
-						n.Broadcast(fmt.Errorf("run JoinAlignNode error: invalid input type but got %[1]T(%[1]v)", d))
-						n.statManager.IncTotalExceptions()
+						e := fmt.Errorf("run JoinAlignNode error: invalid input type but got %[1]T(%[1]v)", d)
+						n.Broadcast(e)
+						n.statManager.IncTotalExceptions(e.Error())
 					}
 				case <-ctx.Done():
 					log.Infoln("Cancelling join align node....")
diff --git a/internal/topo/node/metric/stats_manager.go b/internal/topo/node/metric/stats_manager.go
index 4e45c668..b78f1dc1 100644
--- a/internal/topo/node/metric/stats_manager.go
+++ b/internal/topo/node/metric/stats_manager.go
@@ -22,17 +22,19 @@ import (
 
 const RecordsInTotal = "records_in_total"
 const RecordsOutTotal = "records_out_total"
-const ExceptionsTotal = "exceptions_total"
 const ProcessLatencyUs = "process_latency_us"
 const LastInvocation = "last_invocation"
 const BufferLength = "buffer_length"
+const ExceptionsTotal = "exceptions_total"
+const LastException = "last_exception"
+const LastExceptionTime = "last_exception_time"
 
-var MetricNames = []string{RecordsInTotal, RecordsOutTotal, ExceptionsTotal, ProcessLatencyUs, BufferLength, LastInvocation}
+var MetricNames = []string{RecordsInTotal, RecordsOutTotal, ProcessLatencyUs, BufferLength, LastInvocation, ExceptionsTotal, LastException, LastExceptionTime}
 
 type StatManager interface {
 	IncTotalRecordsIn()
 	IncTotalRecordsOut()
-	IncTotalExceptions()
+	IncTotalExceptions(err string)
 	ProcessTimeStart()
 	ProcessTimeEnd()
 	SetBufferLength(l int64)
@@ -42,12 +44,14 @@ type StatManager interface {
 //The statManager is not thread safe. Make sure it is used in only one instance
 type DefaultStatManager struct {
 	//metrics
-	totalRecordsIn  int64
-	totalRecordsOut int64
-	totalExceptions int64
-	processLatency  int64
-	lastInvocation  time.Time
-	bufferLength    int64
+	totalRecordsIn    int64
+	totalRecordsOut   int64
+	processLatency    int64
+	lastInvocation    time.Time
+	bufferLength      int64
+	totalExceptions   int64
+	lastException     string
+	lastExceptionTime time.Time
 	//configs
 	opType           string //"source", "op", "sink"
 	prefix           string
@@ -86,10 +90,12 @@ func (sm *DefaultStatManager) IncTotalRecordsOut() {
 	sm.totalRecordsOut++
 }
 
-func (sm *DefaultStatManager) IncTotalExceptions() {
+func (sm *DefaultStatManager) IncTotalExceptions(err string) {
 	sm.totalExceptions++
 	var t time.Time
 	sm.processTimeStart = t
+	sm.lastException = err
+	sm.lastExceptionTime = time.Now()
 }
 
 func (sm *DefaultStatManager) ProcessTimeStart() {
@@ -109,14 +115,21 @@ func (sm *DefaultStatManager) SetBufferLength(l int64) {
 
 func (sm *DefaultStatManager) GetMetrics() []interface{} {
 	result := []interface{}{
-		sm.totalRecordsIn, sm.totalRecordsOut, sm.totalExceptions, sm.processLatency, sm.bufferLength,
+		sm.totalRecordsIn,
+		sm.totalRecordsOut,
+		sm.processLatency,
+		sm.bufferLength,
+		0,
+		sm.totalExceptions,
+		sm.lastException,
+		0,
 	}
 
 	if !sm.lastInvocation.IsZero() {
-		result = append(result, sm.lastInvocation.Format("2006-01-02T15:04:05.999999"))
-	} else {
-		result = append(result, 0)
+		result[4] = sm.lastInvocation.Format("2006-01-02T15:04:05.999999")
+	}
+	if !sm.lastExceptionTime.IsZero() {
+		result[7] = sm.lastExceptionTime.Format("2006-01-02T15:04:05.999999")
 	}
-
 	return result
 }
diff --git a/internal/topo/node/metric/stats_prom.go b/internal/topo/node/metric/stats_prom.go
index fdc46a35..1ac3b8df 100644
--- a/internal/topo/node/metric/stats_prom.go
+++ b/internal/topo/node/metric/stats_prom.go
@@ -67,11 +67,9 @@ func (sm *PrometheusStatManager) IncTotalRecordsOut() {
 	sm.pTotalRecordsOut.Inc()
 }
 
-func (sm *PrometheusStatManager) IncTotalExceptions() {
-	sm.totalExceptions++
+func (sm *PrometheusStatManager) IncTotalExceptions(err string) {
 	sm.pTotalExceptions.Inc()
-	var t time.Time
-	sm.processTimeStart = t
+	sm.DefaultStatManager.IncTotalExceptions(err)
 }
 
 func (sm *PrometheusStatManager) ProcessTimeEnd() {
diff --git a/internal/topo/node/operations.go b/internal/topo/node/operations.go
index d948d4b3..0914db4b 100644
--- a/internal/topo/node/operations.go
+++ b/internal/topo/node/operations.go
@@ -136,7 +136,7 @@ func (o *UnaryOperator) doOp(ctx api.StreamContext, errCh chan<- error) {
 			case error:
 				logger.Errorf("Operation %s error: %s", ctx.GetOpId(), val)
 				o.Broadcast(val)
-				stats.IncTotalExceptions()
+				stats.IncTotalExceptions(val.Error())
 				continue
 			default:
 				stats.ProcessTimeEnd()
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index 91727885..260d514d 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -370,7 +370,7 @@ func doCollectData(ctx api.StreamContext, sink api.Sink, outData interface{}, st
 		return nil
 	default:
 		if err := sink.Collect(ctx, outData); err != nil {
-			stats.IncTotalExceptions()
+			stats.IncTotalExceptions(err.Error())
 			return err
 		} else {
 			ctx.GetLogger().Debugf("success")
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 09bc1bc8..10b307f5 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -153,6 +153,11 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 							case err := <-si.errorCh:
 								return err
 							case data := <-buffer.Out:
+								if t, ok := data.(*xsql.ErrorSourceTuple); ok {
+									logger.Errorf("Source %s error: %v", ctx.GetOpId(), t.Error)
+									stats.IncTotalExceptions(t.Error.Error())
+									continue
+								}
 								stats.IncTotalRecordsIn()
 								stats.ProcessTimeStart()
 								tuple := &xsql.Tuple{Emitter: m.name, Message: data.Message(), Timestamp: conf.GetNowInMilli(), Metadata: data.Meta()}
@@ -165,7 +170,7 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 								case error:
 									logger.Errorf("Source %s preprocess error: %s", ctx.GetOpId(), val)
 									m.Broadcast(val)
-									stats.IncTotalExceptions()
+									stats.IncTotalExceptions(val.Error())
 								default:
 									m.Broadcast(val)
 								}
diff --git a/internal/topo/node/watermark.go b/internal/topo/node/watermark.go
index 97f09821..354e51b1 100644
--- a/internal/topo/node/watermark.go
+++ b/internal/topo/node/watermark.go
@@ -203,14 +203,14 @@ func (o *WindowOperator) execEventWindow(ctx api.StreamContext, inputs []*xsql.T
 			}
 			o.statManager.ProcessTimeStart()
 			if !opened {
-				o.statManager.IncTotalExceptions()
+				o.statManager.IncTotalExceptions("input channel closed")
 				break
 			}
 			switch d := item.(type) {
 			case error:
 				o.statManager.IncTotalRecordsIn()
 				o.Broadcast(d)
-				o.statManager.IncTotalExceptions()
+				o.statManager.IncTotalExceptions(d.Error())
 			case xsql.Event:
 				if d.IsWatermark() {
 					watermarkTs := d.GetTimestamp()
@@ -257,8 +257,9 @@ func (o *WindowOperator) execEventWindow(ctx api.StreamContext, inputs []*xsql.T
 				ctx.PutState(WINDOW_INPUTS_KEY, inputs)
 			default:
 				o.statManager.IncTotalRecordsIn()
-				o.Broadcast(fmt.Errorf("run Window error: expect xsql.Event type but got %[1]T(%[1]v)", d))
-				o.statManager.IncTotalExceptions()
+				e := fmt.Errorf("run Window error: expect xsql.Event type but got %[1]T(%[1]v)", d)
+				o.Broadcast(e)
+				o.statManager.IncTotalExceptions(e.Error())
 			}
 		// is cancelling
 		case <-ctx.Done():
diff --git a/internal/topo/node/window_op.go b/internal/topo/node/window_op.go
index 6995dc4c..9b75a040 100644
--- a/internal/topo/node/window_op.go
+++ b/internal/topo/node/window_op.go
@@ -272,13 +272,13 @@ func (o *WindowOperator) execProcessingWindow(ctx api.StreamContext, inputs []*x
 			o.statManager.IncTotalRecordsIn()
 			o.statManager.ProcessTimeStart()
 			if !opened {
-				o.statManager.IncTotalExceptions()
+				o.statManager.IncTotalExceptions("input channel closed")
 				break
 			}
 			switch d := item.(type) {
 			case error:
 				o.Broadcast(d)
-				o.statManager.IncTotalExceptions()
+				o.statManager.IncTotalExceptions(d.Error())
 			case *xsql.Tuple:
 				log.Debugf("Event window receive tuple %s", d.Message)
 				inputs = append(inputs, d)
@@ -332,8 +332,9 @@ func (o *WindowOperator) execProcessingWindow(ctx api.StreamContext, inputs []*x
 				ctx.PutState(WINDOW_INPUTS_KEY, inputs)
 				ctx.PutState(MSG_COUNT_KEY, o.msgCount)
 			default:
-				o.Broadcast(fmt.Errorf("run Window error: expect xsql.Tuple type but got %[1]T(%[1]v)", d))
-				o.statManager.IncTotalExceptions()
+				e := fmt.Errorf("run Window error: expect xsql.Tuple type but got %[1]T(%[1]v)", d)
+				o.Broadcast(e)
+				o.statManager.IncTotalExceptions(e.Error())
 			}
 		case now := <-firstC:
 			log.Debugf("First tick at %v(%d), defined at %d", now, now.UnixMilli(), firstTime)
diff --git a/internal/topo/source/mqtt_source.go b/internal/topo/source/mqtt_source.go
index dbb0ef99..9fe287ef 100644
--- a/internal/topo/source/mqtt_source.go
+++ b/internal/topo/source/mqtt_source.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"path"
@@ -109,50 +110,58 @@ func subscribe(ms *MQTTSource, ctx api.StreamContext, consumer chan<- api.Source
 		return e
 	} else {
 		log.Infof("Successfully subscribed to topic %s.", ms.tpc)
+		var t api.SourceTuple
 		for {
 			select {
 			case <-ctx.Done():
 				log.Infof("Exit subscription to mqtt messagebus topic %s.", ms.tpc)
 				return nil
 			case e1 := <-err:
-				log.Errorf("the subscription to mqtt topic %s have error %s.\n", ms.tpc, e1.Error())
+				t = &xsql.ErrorSourceTuple{
+					Error: fmt.Errorf("the subscription to mqtt topic %s have error %s.\n", ms.tpc, e1.Error()),
+				}
 			case env, ok := <-messages:
 				if !ok { // the source is closed
 					log.Infof("Exit subscription to mqtt messagebus topic %s.", ms.tpc)
 					return nil
 				}
-				msg, ok := env.(pahoMqtt.Message)
-				if !ok {
-					log.Errorf("can not convert interface data to mqtt message %s.", ms.tpc)
-					continue
-				}
-				result, e := ctx.Decode(msg.Payload())
-				//The unmarshal type can only be bool, float64, string, []interface{}, map[string]interface{}, nil
-				if e != nil {
-					log.Errorf("Invalid data format, cannot decode %s to %s format with error %s", string(msg.Payload()), ms.format, e)
-					continue
-				}
-
-				meta := make(map[string]interface{})
-				meta["topic"] = msg.Topic()
-				meta["messageid"] = strconv.Itoa(int(msg.MessageID()))
-
-				if nil != ms.model {
-					sliErr := ms.model.checkType(result, msg.Topic())
-					for _, v := range sliErr {
-						log.Errorf(v)
-					}
-				}
-
-				select {
-				case consumer <- api.NewDefaultSourceTuple(result, meta):
-					log.Debugf("send data to source node")
-				case <-ctx.Done():
-					return nil
-				}
+				t = getTuple(ctx, ms, env)
 			}
+			select {
+			case consumer <- t:
+				log.Debugf("send data to source node")
+			case <-ctx.Done():
+				return nil
+			}
+		}
+	}
+}
+
+func getTuple(ctx api.StreamContext, ms *MQTTSource, env interface{}) api.SourceTuple {
+	msg, ok := env.(pahoMqtt.Message)
+	if !ok { // should never happen
+		return &xsql.ErrorSourceTuple{
+			Error: fmt.Errorf("can not convert interface data to mqtt message %v.", env),
+		}
+	}
+	result, e := ctx.Decode(msg.Payload())
+	//The unmarshal type can only be bool, float64, string, []interface{}, map[string]interface{}, nil
+	if e != nil {
+		return &xsql.ErrorSourceTuple{
+			Error: fmt.Errorf("Invalid data format, cannot decode %s with error %s", string(msg.Payload()), e),
+		}
+	}
+	meta := make(map[string]interface{})
+	meta["topic"] = msg.Topic()
+	meta["messageid"] = strconv.Itoa(int(msg.MessageID()))
+
+	if nil != ms.model {
+		sliErr := ms.model.checkType(result, msg.Topic())
+		for _, v := range sliErr {
+			ctx.GetLogger().Errorf(v)
 		}
 	}
+	return api.NewDefaultSourceTuple(result, meta)
 }
 
 func (ms *MQTTSource) Close(ctx api.StreamContext) error {
diff --git a/internal/xsql/source_tuple.go b/internal/xsql/source_tuple.go
new file mode 100644
index 00000000..03ca8dce
--- /dev/null
+++ b/internal/xsql/source_tuple.go
@@ -0,0 +1,27 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package xsql
+
+type ErrorSourceTuple struct {
+	Error error `json:"error"`
+}
+
+func (t *ErrorSourceTuple) Message() map[string]interface{} {
+	return nil
+}
+
+func (t *ErrorSourceTuple) Meta() map[string]interface{} {
+	return nil
+}


Test patch:
diff --git a/internal/topo/mock/test_sink.go b/internal/topo/mock/test_sink.go
index 1741637d..4d881246 100644
--- a/internal/topo/mock/test_sink.go
+++ b/internal/topo/mock/test_sink.go
@@ -21,11 +21,12 @@ import (
 )
 
 func RunSinkCollect(s api.Sink, data []interface{}) error {
-	ctx := newMockContext("ruleSink", "op1")
+	ctx := NewMockContext("ruleSink", "op1")
 	err := s.Open(ctx)
 	if err != nil {
 		return err
 	}
+	time.Sleep(time.Second)
 	for _, e := range data {
 		err := s.Collect(ctx, e)
 		if err != nil {
diff --git a/internal/topo/mock/test_source.go b/internal/topo/mock/test_source.go
index 680295f7..798eeea9 100644
--- a/internal/topo/mock/test_source.go
+++ b/internal/topo/mock/test_source.go
@@ -31,7 +31,7 @@ func TestSourceOpen(r api.Source, exp []api.SourceTuple, t *testing.T) {
 		count.Store(1)
 		c = 0
 	}
-	ctx, cancel := newMockContext(fmt.Sprintf("rule%d", c), "op1").WithCancel()
+	ctx, cancel := NewMockContext(fmt.Sprintf("rule%d", c), "op1").WithCancel()
 	count.Store(c.(int) + 1)
 	consumer := make(chan api.SourceTuple)
 	errCh := make(chan error)
diff --git a/internal/topo/neuron/neuron_test.go b/internal/topo/neuron/neuron_test.go
index bb5b189b..17483886 100644
--- a/internal/topo/neuron/neuron_test.go
+++ b/internal/topo/neuron/neuron_test.go
@@ -135,6 +135,7 @@ func TestMultiSourceSink(t *testing.T) {
 		},
 	}
 	go func() {
+		time.Sleep(100 * time.Millisecond)
 		err = mock.RunSinkCollect(sin, data)
 		if err != nil {
 			t.Errorf(err.Error())
@@ -172,4 +173,6 @@ func TestMultiSourceSink(t *testing.T) {
 	server.Close()
 	time.Sleep(100 * time.Millisecond)
 	sinkTest(t)
+	sinkConnExpTest(t)
+	connectFailTest(t)
 }
diff --git a/internal/topo/neuron/sink_test.go b/internal/topo/neuron/sink_test.go
index a2a0daa8..c9e281ba 100644
--- a/internal/topo/neuron/sink_test.go
+++ b/internal/topo/neuron/sink_test.go
@@ -81,3 +81,38 @@ func sinkTest(t *testing.T) {
 	}
 	time.Sleep(100 * time.Millisecond)
 }
+
+func sinkConnExpTest(t *testing.T) {
+	s := GetSink()
+	s.Configure(map[string]interface{}{
+		"nodeName":  "test1",
+		"groupName": "grp",
+		"tags":      []string{"temperature", "status"},
+		"raw":       false,
+	})
+	data := []interface{}{
+		map[string]interface{}{
+			"temperature": 22,
+			"humidity":    50,
+			"status":      "green",
+		},
+		map[string]interface{}{
+			"temperature": 25,
+			"humidity":    82,
+			"status":      "wet",
+		},
+		map[string]interface{}{
+			"temperature": 33,
+			"humidity":    60,
+			"status":      "hot",
+		},
+	}
+	expErrStr := "io error: Error publish the tag payload temperature: io error: neuron connection is not established"
+	err := mock.RunSinkCollect(s, data)
+	if err == nil {
+		t.Errorf("should have error")
+		return
+	} else if err.Error() != expErrStr {
+		t.Errorf("error mismatch:\n\nexp=%s\n\ngot=%s\n\n", expErrStr, err.Error())
+	}
+}
diff --git a/internal/topo/neuron/source_test.go b/internal/topo/neuron/source_test.go
index 0c5af3a5..e9b42d1f 100644
--- a/internal/topo/neuron/source_test.go
+++ b/internal/topo/neuron/source_test.go
@@ -15,10 +15,14 @@
 package neuron
 
 import (
+	"fmt"
 	"github.com/lf-edge/ekuiper/internal/topo/mock"
+	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	_ "go.nanomsg.org/mangos/v3/transport/ipc"
+	"reflect"
 	"testing"
+	"time"
 )
 
 func TestRun(t *testing.T) {
@@ -37,3 +41,31 @@ func TestRun(t *testing.T) {
 	defer server.Close()
 	mock.TestSourceOpen(s, exp, t)
 }
+
+func connectFailTest(t *testing.T) {
+	s := GetSource()
+	err := s.Configure("new", nil)
+	if err != nil {
+		t.Errorf(err.Error())
+		return
+	}
+	ctx, cancel := mock.NewMockContext("ruleTestReconnect", "op1").WithCancel()
+	consumer := make(chan api.SourceTuple)
+	errCh := make(chan error)
+	server, _ := mockNeuron(false, false)
+	go s.Open(ctx, consumer, errCh)
+	go func() {
+		select {
+		case err := <-errCh:
+			t.Errorf("received error: %v", err)
+		case tuple := <-consumer:
+			if !reflect.DeepEqual(tuple, &xsql.ErrorSourceTuple{Error: fmt.Errorf("neuron connection detached")}) {
+				t.Errorf("received unexpected tuple: %v", tuple)
+			}
+		}
+		cancel()
+	}()
+	time.Sleep(1 * time.Second)
+	server.Close()
+	time.Sleep(1 * time.Second)
+}