Merge branch 'main' into lovro/write-batch-acks

destination.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/conduitio/conduit-connector-protocol/cpluginv1"
 	"github.com/conduitio/conduit-connector-sdk/internal"
+	"github.com/conduitio/conduit-connector-sdk/internal/csync"
 	"go.uber.org/multierr"
 )
 
@@ -99,7 +100,7 @@ func NewDestinationPlugin(impl Destination) cpluginv1.DestinationPlugin {
 type destinationPluginAdapter struct {
 	impl Destination
 
-	lastPosition *internal.AtomicValueWatcher[Position]
+	lastPosition *csync.ValueWatcher[Position]
 	openCancel   context.CancelFunc
 
 	// write is the chosen write strategy, either single records or batches
@@ -166,7 +167,7 @@ func (a *destinationPluginAdapter) configureWriteStrategy(ctx context.Context, c
 }
 
 func (a *destinationPluginAdapter) Start(ctx context.Context, _ cpluginv1.DestinationStartRequest) (cpluginv1.DestinationStartResponse, error) {
-	a.lastPosition = new(internal.AtomicValueWatcher[Position])
+	a.lastPosition = new(csync.ValueWatcher[Position])
 
 	// detach context, so we can control when it's canceled
 	ctxOpen := internal.DetachContext(ctx)
@@ -205,7 +206,7 @@ func (a *destinationPluginAdapter) Run(ctx context.Context, stream cpluginv1.Des
 		err = a.writeStrategy.Write(ctx, r, func(err error) error {
 			return a.ack(r, err, stream)
 		})
-		a.lastPosition.Store(r.Position)
+		a.lastPosition.Set(r.Position)
 		if err != nil {
 			return err
 		}
@@ -232,14 +233,10 @@ func (a *destinationPluginAdapter) Stop(ctx context.Context, req cpluginv1.Desti
 	// last thing we do is cancel context in Open
 	defer a.openCancel()
 
-	// wait for at most 1 minute
-	waitCtx, cancel := context.WithTimeout(ctx, time.Minute) // TODO make the timeout configurable (https://github.com/ConduitIO/conduit/issues/183)
-	defer cancel()
-
 	// wait for last record to be received
-	err := a.lastPosition.Await(waitCtx, func(val Position) bool {
+	_, err := a.lastPosition.Watch(ctx, func(val Position) bool {
 		return bytes.Equal(val, req.LastPosition)
-	})
+	}, csync.WithTimeout(time.Minute)) // TODO make the timeout configurable (https://github.com/ConduitIO/conduit/issues/183)
 
 	// flush cached records, allow it to take at most 1 minute
 	flushCtx, cancel := context.WithTimeout(ctx, time.Minute) // TODO make the timeout configurable