Fix StreamTarget and KafkaTarget init on flow reuse

[ML-5351](https://jira.iguazeng.com/browse/ML-5351)
mlrun · Dec 26, 2023 · 1cb2ad1 · 1cb2ad1
1 parent 576ffef
commit 1cb2ad1
Showing 1 changed file with 2 additions and 3 deletions.
diff --git a/storey/targets.py b/storey/targets.py
@@ -839,13 +839,13 @@ def __init__(
 
         self._shards = shards
         self._retention_period_hours = retention_period_hours
-        self._initialized = False
 
         self._full_event = full_event
 
     def _init(self):
         Flow._init(self)
         _Writer._init(self)
+        self._initialized = False
         self._worker_exited = False
 
     @staticmethod
@@ -1048,13 +1048,12 @@ def __init__(
         Flow.__init__(self, **kwargs)
         _Writer.__init__(self, columns, infer_columns_from_data, retain_dict=True)
 
-        self._initialized = False
-
         self._full_event = full_event
 
     def _init(self):
         Flow._init(self)
         _Writer._init(self)
+        self._initialized = False
 
     async def _lazy_init(self):
         from kafka import KafkaProducer