Flexible slot name

Add a configuration which allows us to set the name of the replication
slot.

Author: ivanovyordan

tap_postgres/__init__.py

@@ -684,7 +684,8 @@ def main_impl():
                    'filter_dbs' : args.config.get('filter_dbs'),
                    'debug_lsn' : args.config.get('debug_lsn') == 'true',
                    'logical_poll_total_seconds': float(args.config.get('logical_poll_total_seconds', 0)),
-                   'wal2json_message_format': args.config.get('wal2json_message_format')}
+                   'wal2json_message_format': args.config.get('wal2json_message_format'),
+                   'wal2json_slot_name': args.config.get('wal2json_slot_name')}
 
     if args.config.get('ssl') == 'true':
         conn_config['sslmode'] = 'require'

tap_postgres/sync_strategies/logical_replication.py

@@ -379,6 +379,11 @@ def consume_message(streams, state, msg, time_extracted, conn_info, end_lsn, mes
     return state
 
 def locate_replication_slot(conn_info):
+    if conn_info.get('wal2json_slot_name') is not None:
+        slot_name = conn_info["wal2json_slot_name"]
+        LOGGER.info("using pg_replication_slot %s", slot_name)
+        return slot_name
+
     with post_db.open_connection(conn_info, False) as conn:
         with conn.cursor() as cur:
             db_specific_slot = "stitch_{}".format(conn_info['dbname'])