adding DROP MIRROR support for EventHub (#402)

flow/connectors/eventhub/eventhub.go

@@ -350,6 +350,7 @@ func (c *EventHubConnector) SetupNormalizedTables(
 }
 
 func (c *EventHubConnector) SyncFlowCleanup(jobName string) error {
-	// TODO (kaushik): this has to be implemented for DROP PEER support.
-	panic("sync flow cleanup not implemented for event hub")
+	_, err := c.pgMetadata.pool.Exec(c.ctx, fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE",
+		metadataSchema))
+	return err
 }

flow/connectors/postgres/postgres.go

@@ -811,7 +811,8 @@ func (c *PostgresConnector) PullFlowCleanup(jobName string) error {
 		return fmt.Errorf("error dropping publication: %w", err)
 	}
 
-	_, err = pullFlowCleanupTx.Exec(c.ctx, fmt.Sprintf("SELECT pg_drop_replication_slot('%s')", slotName))
+	_, err = pullFlowCleanupTx.Exec(c.ctx, `SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots
+	 WHERE slot_name=$1`, slotName)
 	if err != nil {
 		return fmt.Errorf("error dropping replication slot: %w", err)
 	}

flow/connectors/postgres/postgres_cdc_test.go

@@ -414,8 +414,7 @@ func (suite *PostgresCDCTestSuite) TestErrorForTableNotExist() {
 	}
 
 	err = suite.connector.PullFlowCleanup(nonExistentFlowName)
-	suite.Errorf(err, "error dropping replication slot:"+
-		"ERROR: replication slot \"%s\" does not exist (SQLSTATE 42704)", nonExistentFlowName)
+	suite.Nil(err)
 
 	// creating table and the replication slots for it, and dropping before pull records.
 	_, err = suite.connector.pool.Exec(context.Background(),