bug fixed: the subscription created by a query can be never acked (#1409)

Commelina · web-flow · commit de0d162991df · 2023-05-10T10:21:57.000+08:00
diff --git a/hstream-processing/src/HStream/Processing/Processor.hs b/hstream-processing/src/HStream/Processing/Processor.hs
@@ -221,6 +221,7 @@ runTask statsHolder SourceConnectorWithoutCkp {..} sinkConnector taskBuilder@Tas
                 liftIO $ query_stat_add_total_execute_errors statsHolder (textToCBytes qid) 1
               ]
             liftIO $ query_stat_add_total_output_records statsHolder (textToCBytes qid) 1
+            liftIO $ RIO.putMVar mvar ()
 
 runImmTask ::
   (Ord t, Semigroup t, Aeson.FromJSON t, Aeson.ToJSON t, Typeable t, ChangeLogger h1, Snapshotter h2) =>

Original file line number	Diff line number	Diff line change
`@@ -221,6 +221,7 @@ runTask statsHolder SourceConnectorWithoutCkp {..} sinkConnector taskBuilder@Tas`
`221`	`221`	`liftIO $ query_stat_add_total_execute_errors statsHolder (textToCBytes qid) 1`
`222`	`222`	`]`
`223`	`223`	`liftIO $ query_stat_add_total_output_records statsHolder (textToCBytes qid) 1`
	`224`	`+ liftIO $ RIO.putMVar mvar ()`
`224`	`225`
`225`	`226`	`runImmTask ::`
`226`	`227`	`(Ord t, Semigroup t, Aeson.FromJSON t, Aeson.ToJSON t, Typeable t, ChangeLogger h1, Snapshotter h2) =>`