diff --git a/pkg/stream/integration/helper_test.go b/pkg/stream/integration/helper_test.go
index 35f03a45..6c7062ed 100644
--- a/pkg/stream/integration/helper_test.go
+++ b/pkg/stream/integration/helper_test.go
@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -50,6 +51,8 @@ const (
 	withGeneratedColumn = true
 )
 
+var integrationReplicationSlotCounter uint64
+
 type mockProcessor struct {
 	eventChan   chan *wal.Event
 	skipEventFn func(event *wal.Event) bool
@@ -93,19 +96,39 @@ func (m *mockWebhookServer) close() {
 }
 
 func runStream(t *testing.T, ctx context.Context, cfg *stream.Config) {
-	// start the configured stream listener/processor
+	t.Helper()
+
+	done := make(chan error, 1)
 	go func() {
-		err := stream.Run(ctx, testLogger(), cfg, false, nil)
-		require.NoError(t, err)
+		done <- stream.Run(ctx, testLogger(), cfg, false, nil)
 	}()
+
+	t.Cleanup(func() {
+		select {
+		case err := <-done:
+			require.NoError(t, err)
+		case <-time.After(10 * time.Second):
+			t.Error("timeout waiting for stream to stop")
+		}
+	})
 }
 
 func runSnapshot(t *testing.T, ctx context.Context, cfg *stream.Config) {
-	// start the configured stream listener/processor
+	t.Helper()
+
+	done := make(chan error, 1)
 	go func() {
-		err := stream.Snapshot(ctx, testLogger(), cfg, nil)
-		require.NoError(t, err)
+		done <- stream.Snapshot(ctx, testLogger(), cfg, nil)
 	}()
+
+	t.Cleanup(func() {
+		select {
+		case err := <-done:
+			require.NoError(t, err)
+		case <-time.After(10 * time.Second):
+			t.Error("timeout waiting for snapshot to stop")
+		}
+	})
 }
 
 func initStream(t *testing.T, ctx context.Context, url string) {
@@ -135,17 +158,55 @@ func testLogger() loglib.Logger {
 	}))
 }
 
-func testPostgresListenerCfg() stream.ListenerConfig {
+func testPostgresListenerCfg(t *testing.T) stream.ListenerConfig {
+	t.Helper()
+
+	slotName := initReplicationSlotForTest(t, pgurl)
 	return stream.ListenerConfig{
 		Postgres: &stream.PostgresListenerConfig{
 			URL: pgurl,
 			Replication: pgreplication.Config{
-				PostgresURL: pgurl,
+				PostgresURL:         pgurl,
+				ReplicationSlotName: slotName,
 			},
 		},
 	}
 }
 
+func initReplicationSlotForTest(t *testing.T, url string) string {
+	t.Helper()
+
+	slotName := fmt.Sprintf("pgstream_it_%d_%d", time.Now().UnixNano(), atomic.AddUint64(&integrationReplicationSlotCounter, 1))
+	require.NoError(t, stream.Init(context.Background(), &stream.InitConfig{
+		PostgresURL:         url,
+		ReplicationSlotName: slotName,
+	}))
+
+	t.Cleanup(func() {
+		dropReplicationSlotForTest(t, url, slotName)
+	})
+
+	return slotName
+}
+
+func dropReplicationSlotForTest(t *testing.T, url, slotName string) {
+	t.Helper()
+
+	require.Eventually(t, func() bool {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		conn, err := pglib.NewConn(ctx, url)
+		if err != nil {
+			return false
+		}
+		defer conn.Close(ctx)
+
+		_, err = conn.Exec(ctx, `SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_name = $1`, slotName)
+		return err == nil
+	}, 10*time.Second, 200*time.Millisecond, "replication slot %s was not dropped", slotName)
+}
+
 func testPostgresListenerCfgWithSnapshot(sourceURL, targetURL string, tables []string) stream.ListenerConfig {
 	return stream.ListenerConfig{
 		Postgres: &stream.PostgresListenerConfig{
diff --git a/pkg/stream/integration/pg_kafka_integration_test.go b/pkg/stream/integration/pg_kafka_integration_test.go
index fd51b421..70f06b33 100644
--- a/pkg/stream/integration/pg_kafka_integration_test.go
+++ b/pkg/stream/integration/pg_kafka_integration_test.go
@@ -26,7 +26,7 @@ func Test_PostgresToKafka(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testKafkaProcessorCfg(),
 	}
 
diff --git a/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go b/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go
index 64084e13..cd7032da 100644
--- a/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go
+++ b/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go
@@ -27,7 +27,7 @@ func Test_PostgresToPostgres_BatchCoalesce(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener: testPostgresListenerCfg(),
+		Listener: testPostgresListenerCfg(t),
 		Processor: stream.ProcessorConfig{
 			Postgres: &stream.PostgresProcessorConfig{
 				BatchWriter: postgres.Config{
@@ -195,7 +195,7 @@ func Test_PostgresToPostgres_BatchCoalesce_WithCompositeKey(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener: testPostgresListenerCfg(),
+		Listener: testPostgresListenerCfg(t),
 		Processor: stream.ProcessorConfig{
 			Postgres: &stream.PostgresProcessorConfig{
 				BatchWriter: postgres.Config{
@@ -270,7 +270,7 @@ func Test_PostgresToPostgres_BatchCoalesce_OnConflict(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener: testPostgresListenerCfg(),
+		Listener: testPostgresListenerCfg(t),
 		Processor: stream.ProcessorConfig{
 			Postgres: &stream.PostgresProcessorConfig{
 				BatchWriter: postgres.Config{
diff --git a/pkg/stream/integration/pg_pg_integration_test.go b/pkg/stream/integration/pg_pg_integration_test.go
index ac4c70c5..7bff9758 100644
--- a/pkg/stream/integration/pg_pg_integration_test.go
+++ b/pkg/stream/integration/pg_pg_integration_test.go
@@ -39,7 +39,7 @@ func Test_PostgresToPostgres(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testPostgresProcessorCfg(),
 	}
 
@@ -315,7 +315,7 @@ func Test_PostgresToPostgres_IndexesAndConstraints(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testPostgresProcessorCfg(),
 	}
 
@@ -423,7 +423,7 @@ func Test_PostgresToPostgres_IdentityColumns(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testPostgresProcessorCfg(),
 	}
 
@@ -571,7 +571,7 @@ func Test_PostgresToPostgres_AlwaysIdentityUpdate(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testPostgresProcessorCfg(),
 	}
 
@@ -653,7 +653,7 @@ func Test_PostgresToPostgres_Sequences(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testPostgresProcessorCfg(),
 	}
 
@@ -962,7 +962,7 @@ func Test_PostgresToPostgres_LargeIntegerPrecisionWAL(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testPostgresProcessorCfg(),
 	}
 
diff --git a/pkg/stream/integration/pg_pg_integration_transformer_test.go b/pkg/stream/integration/pg_pg_integration_transformer_test.go
index 52b68087..4d09b27c 100644
--- a/pkg/stream/integration/pg_pg_integration_transformer_test.go
+++ b/pkg/stream/integration/pg_pg_integration_transformer_test.go
@@ -53,7 +53,7 @@ func Test_PostgresToPostgres_Transformer(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testPostgresProcessorCfgWithTransformer(pgurl),
 	}
 
diff --git a/pkg/stream/integration/pg_search_integration_test.go b/pkg/stream/integration/pg_search_integration_test.go
index 1da59153..63fa9b19 100644
--- a/pkg/stream/integration/pg_search_integration_test.go
+++ b/pkg/stream/integration/pg_search_integration_test.go
@@ -131,7 +131,7 @@ func Test_PostgresToSearch(t *testing.T) {
 
 	t.Run("postgres to opensearch", func(t *testing.T) {
 		cfg := &stream.Config{
-			Listener: testPostgresListenerCfg(),
+			Listener: testPostgresListenerCfg(t),
 			Processor: testSearchProcessorCfg(store.Config{
 				OpenSearchURL: opensearchURL,
 			}),
@@ -145,7 +145,7 @@ func Test_PostgresToSearch(t *testing.T) {
 
 	t.Run("postgres to elasticsearch", func(t *testing.T) {
 		cfg := &stream.Config{
-			Listener: testPostgresListenerCfg(),
+			Listener: testPostgresListenerCfg(t),
 			Processor: testSearchProcessorCfg(store.Config{
 				ElasticsearchURL: elasticsearchURL,
 			}),
@@ -162,7 +162,7 @@ func Test_PostgresToSearch(t *testing.T) {
 
 		cfg := testSearchProcessorCfg(store.Config{ElasticsearchURL: elasticsearchURL})
 		cfg.Search.Indexer = search.IndexerConfig{HashDocIDs: true}
-		runStream(t, ctx, &stream.Config{Listener: testPostgresListenerCfg(), Processor: cfg})
+		runStream(t, ctx, &stream.Config{Listener: testPostgresListenerCfg(t), Processor: cfg})
 
 		client, err := elasticsearch.NewClient(elasticsearchURL)
 		require.NoError(t, err)
diff --git a/pkg/stream/integration/pg_webhook_integration_test.go b/pkg/stream/integration/pg_webhook_integration_test.go
index 085cd37b..54c45bd1 100644
--- a/pkg/stream/integration/pg_webhook_integration_test.go
+++ b/pkg/stream/integration/pg_webhook_integration_test.go
@@ -26,7 +26,7 @@ func Test_PostgresToWebhook(t *testing.T) {
 	}
 
 	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
+		Listener:  testPostgresListenerCfg(t),
 		Processor: testWebhookProcessorCfg(),
 	}
 
diff --git a/pkg/stream/integration/setup_test.go b/pkg/stream/integration/setup_test.go
index 3519d19e..b92e536d 100644
--- a/pkg/stream/integration/setup_test.go
+++ b/pkg/stream/integration/setup_test.go
@@ -24,8 +24,8 @@ func TestMain(m *testing.M) {
 
 		if err := stream.Init(ctx, &stream.InitConfig{
 			PostgresURL:               pgurl,
-			ReplicationSlotName:       "",
 			InjectorMigrationsEnabled: true,
+			MigrationsOnly:            true,
 		}); err != nil {
 			log.Fatal(err)
 		}