diff --git a/flow/connectors/eventhub/eventhub.go b/flow/connectors/eventhub/eventhub.go
index c2f3263097..c9a7c7c82c 100644
--- a/flow/connectors/eventhub/eventhub.go
+++ b/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
 }
diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go
index ded4044e6a..20ebe09588 100644
--- a/flow/connectors/postgres/postgres.go
+++ b/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)
 	}
diff --git a/flow/connectors/postgres/postgres_cdc_test.go b/flow/connectors/postgres/postgres_cdc_test.go
index cacc46ec17..ddc9dc3d82 100644
--- a/flow/connectors/postgres/postgres_cdc_test.go
+++ b/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(),