Enable default SDK middleware

.golangci.yml

@@ -1,5 +1,5 @@
 run:
-  timeout: 5m 
+  timeout: 5m
 
 linters-settings:
   nolintlint:
@@ -50,7 +50,6 @@ linters:
     - errorlint
     - exhaustive
     # - exhaustivestruct
-    - exportloopref
     # - forbidigo
     # - forcetypeassert
     - gochecknoinits
@@ -111,4 +110,4 @@ linters:
     - testableexamples
     - unconvert
     - usestdlibvars
-    - zerologlint
+    - zerologlint

destination/destination.go

@@ -28,7 +28,7 @@ import (
 )
 
 func NewDestination() sdk.Destination {
-	return &Destination{}
+	return sdk.DestinationWithMiddleware(&Destination{}, sdk.DefaultDestinationMiddleware()...)
 }
 
 type Destination struct {

test/test_util.go

@@ -0,0 +1,41 @@
+// Copyright © 2024 Meroxa, Inc. and Miquido
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package test
+
+import (
+	"context"
+	"fmt"
+
+	esDestination "github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch"
+	sdk "github.com/conduitio/conduit-connector-sdk"
+)
+
+func GetClient(cfgMap map[string]string) (elasticsearch.Client, error) {
+	ctx := context.Background()
+	cfg := esDestination.Config{}
+
+	err := sdk.Util.ParseConfig(ctx, cfgMap, &cfg, esDestination.NewDestination().Parameters())
+	if err != nil {
+		return nil, fmt.Errorf("failed parsing config: %w", err)
+	}
+
+	client, err := elasticsearch.NewClient(cfg.Version, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed creating client: %w", err)
+	}
+
+	return client, nil
+}

test/v5/acceptance_test.go

@@ -24,6 +24,7 @@ import (
 	esDestination "github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch"
 	v5 "github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/v5"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/test"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"go.uber.org/goleak"
@@ -60,8 +61,6 @@ func (d *CustomConfigurableAcceptanceTestDriver) ReadFromDestination(_ *testing.
 }
 
 func TestAcceptance(t *testing.T) {
-	var dest *esDestination.Destination
-
 	destinationConfig := map[string]string{
 		esDestination.ConfigVersion:  elasticsearch.Version5,
 		esDestination.ConfigHost:     "http://127.0.0.1:9200",
@@ -70,25 +69,24 @@ func TestAcceptance(t *testing.T) {
 		esDestination.ConfigBulkSize: "100",
 	}
 
+	client, err := test.GetClient(destinationConfig)
+	if err != nil {
+		t.Logf("error creating client: %v", err)
+	}
+
 	sdk.AcceptanceTest(t, &CustomConfigurableAcceptanceTestDriver{
 		ConfigurableAcceptanceTestDriver: sdk.ConfigurableAcceptanceTestDriver{
 			Config: sdk.ConfigurableAcceptanceTestDriverConfig{
 				Connector: sdk.Connector{
 					NewSpecification: es.Specification,
-
-					NewSource: nil,
-
-					NewDestination: func() sdk.Destination {
-						dest = esDestination.NewDestination().(*esDestination.Destination)
-
-						return dest
-					},
+					NewSource:        nil,
+					NewDestination:   esDestination.NewDestination,
 				},
 
 				DestinationConfig: destinationConfig,
 
 				AfterTest: func(_ *testing.T) {
-					if client := dest.GetClient(); client != nil {
+					if client != nil {
 						assertIndexIsDeleted(
 							client.(*v5.Client).GetClient(),
 							destinationConfig[esDestination.ConfigIndex],

test/v5/connector_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch"
 	v5 "github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/v5"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/test"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	esV5 "github.com/elastic/go-elasticsearch/v5"
@@ -35,7 +36,7 @@ import (
 
 func TestOperationsWithSmallestBulkSize(t *testing.T) {
 	fakerInstance := faker.New()
-	dest := destination.NewDestination().(*destination.Destination)
+	dest := destination.NewDestination()
 
 	cfgRaw := map[string]string{
 		destination.ConfigVersion:  elasticsearch.Version5,
@@ -45,11 +46,15 @@ func TestOperationsWithSmallestBulkSize(t *testing.T) {
 		destination.ConfigBulkSize: "1",
 	}
 
+	client, err := test.GetClient(cfgRaw)
+	if err != nil {
+		t.Logf("failed to create elasticsearch client: %v", err)
+	}
+	esClient := client.(*v5.Client).GetClient()
+
 	require.NoError(t, dest.Configure(context.Background(), cfgRaw))
 	require.NoError(t, dest.Open(context.Background()))
 
-	esClient := dest.GetClient().(*v5.Client).GetClient()
-
 	require.True(t, assertIndexIsDeleted(esClient, "users"))
 
 	t.Cleanup(func() {
@@ -263,7 +268,7 @@ func TestOperationsWithSmallestBulkSize(t *testing.T) {
 
 func TestOperationsWithBiggerBulkSize(t *testing.T) {
 	fakerInstance := faker.New()
-	dest := destination.NewDestination().(*destination.Destination)
+	dest := destination.NewDestination()
 
 	cfgRaw := map[string]string{
 		destination.ConfigVersion:  elasticsearch.Version5,
@@ -273,11 +278,15 @@ func TestOperationsWithBiggerBulkSize(t *testing.T) {
 		destination.ConfigBulkSize: "3",
 	}
 
+	client, err := test.GetClient(cfgRaw)
+	if err != nil {
+		t.Logf("failed to create elasticsearch client: %v", err)
+	}
+	esClient := client.(*v5.Client).GetClient()
+
 	require.NoError(t, dest.Configure(context.Background(), cfgRaw))
 	require.NoError(t, dest.Open(context.Background()))
 
-	esClient := dest.GetClient().(*v5.Client).GetClient()
-
 	require.True(t, assertIndexIsDeleted(esClient, "users"))
 
 	t.Cleanup(func() {

test/v6/acceptance_test.go

@@ -24,6 +24,7 @@ import (
 	esDestination "github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch"
 	v6 "github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/v6"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/test"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"go.uber.org/goleak"
@@ -60,8 +61,6 @@ func (d *CustomConfigurableAcceptanceTestDriver) ReadFromDestination(_ *testing.
 }
 
 func TestAcceptance(t *testing.T) {
-	var dest *esDestination.Destination
-
 	destinationConfig := map[string]string{
 		esDestination.ConfigVersion:  elasticsearch.Version6,
 		esDestination.ConfigHost:     "http://127.0.0.1:9200",
@@ -70,6 +69,11 @@ func TestAcceptance(t *testing.T) {
 		esDestination.ConfigBulkSize: "100",
 	}
 
+	client, err := test.GetClient(destinationConfig)
+	if err != nil {
+		t.Logf("error creating client: %v", err)
+	}
+
 	sdk.AcceptanceTest(t, &CustomConfigurableAcceptanceTestDriver{
 		ConfigurableAcceptanceTestDriver: sdk.ConfigurableAcceptanceTestDriver{
 			Config: sdk.ConfigurableAcceptanceTestDriverConfig{
@@ -78,17 +82,13 @@ func TestAcceptance(t *testing.T) {
 
 					NewSource: nil,
 
-					NewDestination: func() sdk.Destination {
-						dest = esDestination.NewDestination().(*esDestination.Destination)
-
-						return dest
-					},
+					NewDestination: esDestination.NewDestination,
 				},
 
 				DestinationConfig: destinationConfig,
 
 				AfterTest: func(_ *testing.T) {
-					if client := dest.GetClient(); client != nil {
+					if client != nil {
 						assertIndexIsDeleted(
 							client.(*v6.Client).GetClient(),
 							destinationConfig[esDestination.ConfigIndex],

test/v6/connector_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch"
 	v6 "github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/v6"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/test"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	esV6 "github.com/elastic/go-elasticsearch/v6"
@@ -35,7 +36,7 @@ import (
 
 func TestOperationsWithSmallestBulkSize(t *testing.T) {
 	fakerInstance := faker.New()
-	dest := destination.NewDestination().(*destination.Destination)
+	dest := destination.NewDestination()
 
 	cfgRaw := map[string]string{
 		destination.ConfigVersion:  elasticsearch.Version6,
@@ -48,7 +49,11 @@ func TestOperationsWithSmallestBulkSize(t *testing.T) {
 	require.NoError(t, dest.Configure(context.Background(), cfgRaw))
 	require.NoError(t, dest.Open(context.Background()))
 
-	esClient := dest.GetClient().(*v6.Client).GetClient()
+	client, err := test.GetClient(cfgRaw)
+	if err != nil {
+		t.Logf("failed to create elasticsearch client: %v", err)
+	}
+	esClient := client.(*v6.Client).GetClient()
 
 	require.True(t, assertIndexIsDeleted(esClient, "users"))
 
@@ -263,7 +268,7 @@ func TestOperationsWithSmallestBulkSize(t *testing.T) {
 
 func TestOperationsWithBiggerBulkSize(t *testing.T) {
 	fakerInstance := faker.New()
-	dest := destination.NewDestination().(*destination.Destination)
+	dest := destination.NewDestination()
 
 	cfgRaw := map[string]string{
 		destination.ConfigVersion:  elasticsearch.Version6,
@@ -276,7 +281,11 @@ func TestOperationsWithBiggerBulkSize(t *testing.T) {
 	require.NoError(t, dest.Configure(context.Background(), cfgRaw))
 	require.NoError(t, dest.Open(context.Background()))
 
-	esClient := dest.GetClient().(*v6.Client).GetClient()
+	client, err := test.GetClient(cfgRaw)
+	if err != nil {
+		t.Logf("failed to create elasticsearch client: %v", err)
+	}
+	esClient := client.(*v6.Client).GetClient()
 
 	require.True(t, assertIndexIsDeleted(esClient, "users"))
 

test/v7/acceptance_test.go

@@ -24,6 +24,7 @@ import (
 	esDestination "github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch"
 	v7 "github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/v7"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/test"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"go.uber.org/goleak"
@@ -60,15 +61,18 @@ func (d *CustomConfigurableAcceptanceTestDriver) ReadFromDestination(_ *testing.
 }
 
 func TestAcceptance(t *testing.T) {
-	var dest *esDestination.Destination
-
 	destinationConfig := map[string]string{
 		esDestination.ConfigVersion:  elasticsearch.Version7,
 		esDestination.ConfigHost:     "http://127.0.0.1:9200",
 		esDestination.ConfigIndex:    "acceptance_idx",
 		esDestination.ConfigBulkSize: "100",
 	}
 
+	client, err := test.GetClient(destinationConfig)
+	if err != nil {
+		t.Logf("failed to create elasticsearch client: %v", err)
+	}
+
 	sdk.AcceptanceTest(t, &CustomConfigurableAcceptanceTestDriver{
 		ConfigurableAcceptanceTestDriver: sdk.ConfigurableAcceptanceTestDriver{
 			Config: sdk.ConfigurableAcceptanceTestDriverConfig{
@@ -77,17 +81,13 @@ func TestAcceptance(t *testing.T) {
 
 					NewSource: nil,
 
-					NewDestination: func() sdk.Destination {
-						dest = esDestination.NewDestination().(*esDestination.Destination)
-
-						return dest
-					},
+					NewDestination: esDestination.NewDestination,
 				},
 
 				DestinationConfig: destinationConfig,
 
 				AfterTest: func(_ *testing.T) {
-					if client := dest.GetClient(); client != nil {
+					if client != nil {
 						assertIndexIsDeleted(
 							client.(*v7.Client).GetClient(),
 							destinationConfig[esDestination.ConfigIndex],

test/v7/connector_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/destination"
 	"github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch"
 	v7 "github.com/conduitio-labs/conduit-connector-elasticsearch/internal/elasticsearch/v7"
+	"github.com/conduitio-labs/conduit-connector-elasticsearch/test"
 	"github.com/conduitio/conduit-commons/opencdc"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	esV7 "github.com/elastic/go-elasticsearch/v7"
@@ -35,7 +36,7 @@ import (
 
 func TestOperationsWithSmallestBulkSize(t *testing.T) {
 	fakerInstance := faker.New()
-	dest := destination.NewDestination().(*destination.Destination)
+	dest := destination.NewDestination()
 
 	cfgRaw := map[string]string{
 		destination.ConfigVersion:  elasticsearch.Version7,
@@ -48,7 +49,11 @@ func TestOperationsWithSmallestBulkSize(t *testing.T) {
 	require.NoError(t, dest.Configure(context.Background(), cfgRaw))
 	require.NoError(t, dest.Open(context.Background()))
 
-	esClient := dest.GetClient().(*v7.Client).GetClient()
+	client, err := test.GetClient(cfgRaw)
+	if err != nil {
+		t.Logf("failed to create elasticsearch client: %v", err)
+	}
+	esClient := client.(*v7.Client).GetClient()
 
 	require.True(t, assertIndexIsDeleted(esClient, "users"))
 
@@ -263,7 +268,7 @@ func TestOperationsWithSmallestBulkSize(t *testing.T) {
 
 func TestOperationsWithBiggerBulkSize(t *testing.T) {
 	fakerInstance := faker.New()
-	dest := destination.NewDestination().(*destination.Destination)
+	dest := destination.NewDestination()
 
 	cfgRaw := map[string]string{
 		destination.ConfigVersion:  elasticsearch.Version7,
@@ -276,7 +281,11 @@ func TestOperationsWithBiggerBulkSize(t *testing.T) {
 	require.NoError(t, dest.Configure(context.Background(), cfgRaw))
 	require.NoError(t, dest.Open(context.Background()))
 
-	esClient := dest.GetClient().(*v7.Client).GetClient()
+	client, err := test.GetClient(cfgRaw)
+	if err != nil {
+		t.Logf("failed to create elasticsearch client: %v", err)
+	}
+	esClient := client.(*v7.Client).GetClient()
 
 	require.True(t, assertIndexIsDeleted(esClient, "users"))