fix: cherry-pic fix to master-2.2 (#3853)

Signed-off-by: Jiyong Huang <huangjy@emqx.io>
Signed-off-by: Song Gao <disxiaofei@163.com>
Co-authored-by: Jiyong Huang <huangjy@emqx.io>

Author: Yisaer

extensions/impl/sql/source.go

@@ -330,6 +330,11 @@ func buildScanValueByColumnType(ctx api.StreamContext, colName, colType string,
 			return &sql.NullInt64{}
 		}
 		return new(int64)
+	case "TIMESTAMP":
+		if nullable {
+			return &sql.NullTime{}
+		}
+		return new(time.Time)
 	default:
 		ctx.GetLogger().Debugf("sql source meet column %v unknown columnType:%v", colName, colType)
 		return nil

internal/server/schema_init.go

@@ -77,7 +77,7 @@ func schemasHandler(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		if err = sch.Validate(); err != nil {
-			handleError(w, nil, "Invalid body", logger)
+			handleError(w, err, "Invalid body", logger)
 			return
 		}
 		err = schema.Register(sch)
@@ -120,7 +120,6 @@ func schemaHandler(w http.ResponseWriter, r *http.Request) {
 			handleError(w, err, "schema update command error", logger)
 			return
 		}
-		w.WriteHeader(http.StatusOK)
 	case http.MethodPut:
 		sch := &schema.Info{Type: def.SchemaType(st), Name: name}
 		err := json.NewDecoder(r.Body).Decode(sch)

internal/topo/node/source_node.go

@@ -258,7 +258,7 @@ func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
 	defer func() {
 		m.s.Close(ctx)
 		m.Close()
-		if m.notifySub {
+		if m.notifySub && sig.Ctrl != nil {
 			sig.Ctrl.Rem(m.name)
 		}
 	}()
@@ -282,7 +282,7 @@ func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
 		if err != nil {
 			return err
 		}
-		if m.notifySub {
+		if m.notifySub && sig.Ctrl != nil {
 			sig.Ctrl.Add(m.name)
 		}
 		return nil

sdk/go/runtime/sink.go

@@ -16,7 +16,6 @@ package runtime
 
 import (
 	context2 "context"
-	"encoding/json"
 	"fmt"
 
 	"go.nanomsg.org/mangos/v3"
@@ -90,16 +89,17 @@ func (s *sinkRuntime) run() {
 		if err != nil {
 			s.ctx.GetLogger().Errorf("collect error: %s", err.Error())
 		}
-		r := &ackResponse{}
-		if err != nil {
-			r.Error = err.Error()
-		}
-		data, _ := json.Marshal(r)
-		if err := s.ackCh.Send(data); err != nil {
-			s.ctx.GetLogger().Errorf("ack error: %s", err.Error())
-			_ = s.stop()
-			return
-		}
+		// temporary remove golang plugin sink ack
+		//r := &ackResponse{}
+		//if err != nil {
+		//	r.Error = err.Error()
+		//}
+		//data, _ := json.Marshal(r)
+		//if err := s.ackCh.Send(data); err != nil {
+		//	s.ctx.GetLogger().Errorf("ack error: %s", err.Error())
+		//	_ = s.stop()
+		//	return
+		//}
 	}
 }