fix(connector): add fixed-connector-image options

conf/hstream.yaml

@@ -119,6 +119,12 @@ hserver:
     # so the tasks-network should be the network that can connect to HStreamDB and external systems.
     tasks-network: host
     extra-docker-args: "-m 4g"
+
+    # when restarting a connector, whether use fixed connector image(if images of config file was updated),
+    # true -> read image from metastore
+    # false -> read image from config file
+    fixed-connector-image: true
+
     # source images
     source-images:
       mysql: hstreamdb/source-mysql:latest

hstream-io/HStream/IO/Types.hs

@@ -60,11 +60,12 @@ data IOTaskStatus
   deriving (Show, Eq, Generic, J.FromJSON, J.ToJSON)
 
 data IOOptions = IOOptions
-  { optTasksNetwork    :: T.Text
-  , optTasksPath       :: T.Text
-  , optSourceImages    :: HM.HashMap T.Text T.Text
-  , optSinkImages      :: HM.HashMap T.Text T.Text
-  , optExtraDockerArgs :: T.Text
+  { optTasksNetwork        :: T.Text
+  , optTasksPath           :: T.Text
+  , optSourceImages        :: HM.HashMap T.Text T.Text
+  , optSinkImages          :: HM.HashMap T.Text T.Text
+  , optExtraDockerArgs     :: T.Text
+  , optFixedConnectorImage :: Bool
   } deriving (Show, Eq)
 
 type TaskProcess = TP.Process IO.Handle IO.Handle ()

hstream-io/HStream/IO/Worker.hs

@@ -176,7 +176,11 @@ recoverTask worker@Worker{..} name = do
     Nothing -> throwIO $ HE.ConnectorNotFound name
     Just (taskId, TaskMeta{taskInfoMeta=taskInfo@TaskInfo{..}}) -> do
       let newConnCfg = J.insert "hstream" (J.toJSON hsConfig) connectorConfig
-      createIOTaskFromTaskInfo worker taskId taskInfo{connectorConfig=newConnCfg} options True False False
+          newImage = if options.optFixedConnectorImage
+                     then taskConfig.tcImage
+                     else makeImage taskType taskTarget options
+          newTaskConfig = taskConfig{tcImage=newImage}
+      createIOTaskFromTaskInfo worker taskId taskInfo{connectorConfig=newConnCfg, taskConfig=newTaskConfig} options True False False
 
 -- update config and restart
 alterConnectorConfig :: Worker -> T.Text -> T.Text -> IO ()

hstream/src/HStream/Server/Config.hs

@@ -238,6 +238,7 @@ parseJSONToOptions CliOptions{..} obj = do
   nodeSourceImages <- nodeIOCfg .:? "source-images" .!= HM.empty
   nodeSinkImages <- nodeIOCfg .:? "sink-images" .!= HM.empty
   optExtraDockerArgs <- nodeIOCfg .:? "extra-docker-args" .!= ""
+  optFixedConnectorImage <- nodeIOCfg .:? "fixed-connector-image" .!= True
   (optSourceImages, optSinkImages) <- foldrM
         (\img (ss, sk) -> do
           -- "source mysql IMAGE" -> ("source" "mysq" "IMAGE")