Add disable eventhandler option

doc/config.md

@@ -9,6 +9,11 @@ If you set this parameter to `True`, SB-OSC will skip the bulk import stage and
 ### disable_apply_dml_events
 If you set this parameter to `True`, SB-OSC will pause before `apply_dml_events` stage. This is useful when you have additional steps to perform manually before applying DML events.
 
+### disable_eventhandler
+If you set this parameter to `True`, SB-OSC will disable eventhandler, which means it will not process binlog events. Only bulk import will be performed.  
+
+After `bulk_import_validation` stage it will move directly to `done` stage. So, `add_index` stage will be skipped since `apply_dml_events` stage will not be executed.
+
 
 ## Chunk
 ### max_chunk_count & min_chunk_size

src/config/config.py

@@ -84,6 +84,7 @@ class Config:
     USE_BATCH_SIZE_MULTIPLIER = False
 
     # EventHandler config
+    DISABLE_EVENTHANDLER = False
     EVENTHANDLER_THREAD_COUNT = 4
     EVENTHANDLER_THREAD_TIMEOUT_IN_SECONDS = 300
     INIT_BINLOG_FILE: str = None

src/sbosc/controller/controller.py

@@ -52,7 +52,6 @@ def start(self):
             if action:
                 action()
 
-            # TODO: Add Redis data validation if needed
             time.sleep(self.interval)
 
         # Close db connection
@@ -166,6 +165,10 @@ def validate_bulk_import(self):
                     self.redis_data.set_current_stage(Stage.BULK_IMPORT_VALIDATION_FAILED)
                     self.slack.send_message(message="Bulk import validation failed", color="danger")
                 else:
+                    if not config.DISABLE_EVENTHANDLER:
+                        self.redis_data.set_current_stage(Stage.APPLY_DML_EVENTS)
+                    else:
+                        self.redis_data.set_current_stage(Stage.DONE)
                     self.redis_data.set_current_stage(Stage.APPLY_DML_EVENTS)
                     self.slack.send_message(message="Bulk import validation succeeded", color="good")
             except StopFlagSet:

src/sbosc/controller/initializer.py

@@ -196,7 +196,10 @@ def fetch_metadata(self, redis_data):
             self.logger.info("Saved total rows to Redis")
 
         metadata.start_datetime = datetime.now()
-        redis_data.set_current_stage(Stage.START_EVENT_HANDLER)
+        if not config.DISABLE_EVENTHANDLER:
+            redis_data.set_current_stage(Stage.START_EVENT_HANDLER)
+        else:
+            redis_data.set_current_stage(Stage.BULK_IMPORT_CHUNK_CREATION)
 
     def init_migration(self):
         if not self.check_database_setup():

src/sbosc/eventhandler/eventhandler.py

@@ -188,7 +188,7 @@ def start(self):
         self.logger.info('Starting event handler')
         while not self.stop_flag:
             current_stage = self.redis_data.current_stage
-            if Stage.DONE > current_stage >= Stage.START_EVENT_HANDLER:
+            if Stage.DONE > current_stage >= Stage.START_EVENT_HANDLER and not config.DISABLE_EVENTHANDLER:
                 if self.log_file is None or self.log_pos is None:
                     self.logger.info('Initializing event handler')
                     self.init_event_handler()

src/sbosc/monitor/monitor.py

@@ -310,6 +310,9 @@ def check_migration_status(self):
             bulk_import_progress = inserted_rows / self.redis_data.metadata.max_pk * 100
             self.metric_sender.submit('sb_osc_bulk_import_progress', bulk_import_progress)
 
+        if config.DISABLE_EVENTHANDLER:
+            return
+
         self.submit_event_handler_timestamps()
 
         # remaining_binlog_size