diff --git a/internal/impl/snowflake/output_snowflake_streaming.go b/internal/impl/snowflake/output_snowflake_streaming.go index ed94eb608c..cf9ad5a330 100644 --- a/internal/impl/snowflake/output_snowflake_streaming.go +++ b/internal/impl/snowflake/output_snowflake_streaming.go @@ -583,7 +583,6 @@ const SnowflakeClientResourceForTesting snowflakeClientForTesting = "SnowflakeCl type snowpipeStreamingOutput struct { initStatementsFn func(context.Context, *streaming.SnowflakeRestClient) error - client *streaming.SnowflakeServiceClient restClient *streaming.SnowflakeRestClient mapping *bloblang.Executor logger *service.Logger @@ -695,7 +694,6 @@ func (o *snowpipeStreamingOutput) Close(ctx context.Context) error { return err } o.restClient.Close() - o.client.Close() return nil } @@ -785,6 +783,7 @@ func (o *snowpipePooledOutput) WriteBatch(ctx context.Context, batch service.Mes func (o *snowpipePooledOutput) Close(ctx context.Context) error { o.channelPool.Reset() + o.client.Close() return nil } @@ -872,6 +871,7 @@ func (o *snowpipeIndexedOutput) WriteBatch(ctx context.Context, batch service.Me func (o *snowpipeIndexedOutput) Close(ctx context.Context) error { o.channelPool.Reset() + o.client.Close() return nil }