Repository: lf-edge/ekuiper
Pull request #1354 (2022-08-10 03:14:33)

fix(ci): upgrade build base image to go 1.18.5
Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>


Patch:
diff --git a/.github/dockerfile/Dockerfile.centos b/.github/dockerfile/Dockerfile.centos
index 34e62a72..6a880359 100644
--- a/.github/dockerfile/Dockerfile.centos
+++ b/.github/dockerfile/Dockerfile.centos
@@ -16,7 +16,7 @@ FROM centos:7
 
 RUN yum install -y make wget git which rpm-build gcc
 
-ARG GO_VERSION=1.18
+ARG GO_VERSION=1.18.5
 
 RUN case $(arch) in \
         x86_64) \
diff --git a/.github/dockerfile/Dockerfile.debian b/.github/dockerfile/Dockerfile.debian
index d19ea355..91b53b95 100644
--- a/.github/dockerfile/Dockerfile.debian
+++ b/.github/dockerfile/Dockerfile.debian
@@ -12,12 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-ARG GO_VERSION=1.18
-FROM golang:${GO_VERSION} AS builder
+FROM debian:stretch
+
+ARG GO_VERSION=1.18.5
+
+RUN apt update \
+    && apt install -y zip upx pkg-config libczmq-dev build-essential debhelper jq zip
+
+RUN wget --no-verbose https://golang.org/dl/go$GO_VERSION.linux-$(dpkg --print-architecture).tar.gz \
+    && tar -C /usr/local -xzf go$GO_VERSION.linux-$(dpkg --print-architecture).tar.gz
+
+ENV PATH=$PATH:/usr/local/go/bin
 
 COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN apt update \
-    && apt install -y zip upx pkg-config libczmq-dev build-essential debhelper jq zip
+ENV KUIPER_SOURCE=/go/kuiper
diff --git a/.github/workflows/build_base_image.yaml b/.github/workflows/build_base_image.yaml
index 2f9ef20a..c71ddca0 100644
--- a/.github/workflows/build_base_image.yaml
+++ b/.github/workflows/build_base_image.yaml
@@ -21,7 +21,7 @@ jobs:
           - debian
           - centos
         golang:
-          - 1.18
+          - 1.18.5
 
     steps:
       - uses: actions/checkout@v2
diff --git a/internal/topo/connection/clients/mqtt/mqtt.go b/internal/topo/connection/clients/mqtt/mqtt.go
index a3491d9c..7ebfeb40 100644
--- a/internal/topo/connection/clients/mqtt/mqtt.go
+++ b/internal/topo/connection/clients/mqtt/mqtt.go
@@ -73,10 +73,10 @@ func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
 	} else {
 		ms.clientid = cfg.ClientId
 	}
-
-	ms.pVersion = 3
-	if cfg.PVersion == "3.1.1" {
-		ms.pVersion = 4
+	// Default to MQTT 3.1.1 or NanoMQ cannot connect
+	ms.pVersion = 4
+	if cfg.PVersion == "3.1" {
+		ms.pVersion = 3
 	}
 
 	tlsOpts := cert.TlsConfigurationOptions{
@@ -98,8 +98,10 @@ func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
 }
 
 func (ms *MQTTClient) Connect(connHandler MQTT.OnConnectHandler, lostHandler MQTT.ConnectionLostHandler) error {
-
-	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(ms.pVersion)
+	if conf.Config.Basic.Debug {
+		MQTT.DEBUG = conf.Log
+	}
+	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(4)
 
 	opts = opts.SetTLSConfig(ms.tls)
 
diff --git a/internal/topo/node/source_pool.go b/internal/topo/node/source_pool.go
index 15e65c50..c69a7831 100644
--- a/internal/topo/node/source_pool.go
+++ b/internal/topo/node/source_pool.go
@@ -24,6 +24,7 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 	"sync"
+	"time"
 )
 
 //// Package vars and funcs
@@ -273,15 +274,27 @@ func (ss *sourceSingleton) broadcastError(err error) {
 }
 
 func (ss *sourceSingleton) attach(instanceKey string, bl int) error {
-	ss.Lock()
-	defer ss.Unlock()
-	if _, ok := ss.outputs[instanceKey]; !ok {
-		ss.outputs[instanceKey] = newSourceInstanceChannels(bl)
-	} else {
-		// should not happen
-		return fmt.Errorf("fail to attach source instance, already has an output of the same key %s", instanceKey)
+	retry := 10
+	var err error
+	// retry multiple times in case the detach is still in progress
+	for i := 0; i < retry; i++ {
+		err = func() error {
+			ss.Lock()
+			defer ss.Unlock()
+			if _, ok := ss.outputs[instanceKey]; !ok {
+				ss.outputs[instanceKey] = newSourceInstanceChannels(bl)
+			} else {
+				// should not happen
+				return fmt.Errorf("fail to attach source instance, already has an output of the same key %s", instanceKey)
+			}
+			return nil
+		}()
+		if err == nil {
+			return nil
+		}
+		time.Sleep(time.Millisecond * 100)
 	}
-	return nil
+	return err
 }
 
 // detach Detach an instance and return if the singleton is ended


Test patch:
diff --git a/internal/topo/connection/clients/mqtt/client_mqtt_test.go b/internal/topo/connection/clients/mqtt/client_mqtt_test.go
index 1899a494..ce73f2f3 100644
--- a/internal/topo/connection/clients/mqtt/client_mqtt_test.go
+++ b/internal/topo/connection/clients/mqtt/client_mqtt_test.go
@@ -131,7 +131,7 @@ func TestMQTTClient_CfgResult(t *testing.T) {
 	if !reflect.DeepEqual("clientid", ms.clientid) {
 		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.clientid)
 	}
-	if !reflect.DeepEqual(uint(3), ms.pVersion) {
+	if !reflect.DeepEqual(uint(4), ms.pVersion) {
 		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.pVersion)
 	}