pr feedback

Author: hariso

pkg/connector/destination.go

@@ -234,7 +234,7 @@ func (d *destination) Write(ctx context.Context, r record.Record) error {
 		return err
 	}
 
-	d.inspector.Send(r)
+	d.inspector.Send(ctx, r)
 	err = d.plugin.Write(ctx, r)
 	if err != nil {
 		return cerrors.Errorf("error writing record: %w", err)

pkg/inspector/inspector.go

@@ -27,28 +27,26 @@ import (
 // 1. a way to send records to it asynchronously
 // 2. a way to know if it's closed or not
 type Session struct {
-	C           chan record.Record
-	CloseReason error
+	C chan record.Record
 
 	logger  log.CtxLogger
 	onClose func()
 }
 
-func (s *Session) Close(err error) {
+func (s *Session) Close() {
 	s.onClose()
 	close(s.C)
-	s.CloseReason = err
 }
 
 // send sends a record to the session's channel.
 // If the channel has already reached its capacity,
 // the record will be ignored.
-func (s *Session) send(r record.Record) {
+func (s *Session) send(ctx context.Context, r record.Record) {
 	select {
 	case s.C <- r:
 	default:
 		s.logger.
-			Warn(context.Background()).
+			Warn(ctx).
 			Msg("session buffer full, record will be dropped")
 	}
 }
@@ -77,7 +75,7 @@ func New(logger log.CtxLogger, bufferSize int) *Inspector {
 
 // Send sends the given record to all registered sessions.
 // The method does not wait for consumers to get the records.
-func (i *Inspector) Send(r record.Record) {
+func (i *Inspector) Send(ctx context.Context, r record.Record) {
 	// copy metadata, to prevent issues when concurrently accessing the metadata
 	var meta record.Metadata
 	if len(r.Metadata) != 0 {
@@ -93,7 +91,7 @@ func (i *Inspector) Send(r record.Record) {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 	for _, s := range i.sessions {
-		s.send(record.Record{
+		s.send(ctx, record.Record{
 			Position:  r.Position,
 			Operation: r.Operation,
 			Metadata:  meta,
@@ -114,7 +112,10 @@ func (i *Inspector) NewSession(ctx context.Context) *Session {
 	}
 	go func() {
 		<-ctx.Done()
-		s.Close(ctx.Err())
+		s.logger.
+			Info(context.Background()).
+			Msgf("context canceled: %v", ctx.Err())
+		s.Close()
 	}()
 
 	i.lock.Lock()

pkg/inspector/inspector_benchmark_test.go

@@ -27,6 +27,6 @@ func BenchmarkInspector_SingleSession_Send(b *testing.B) {
 	ins.NewSession(context.Background())
 
 	for i := 0; i < b.N; i++ {
-		ins.Send(record.Record{Position: record.Position("test-pos")})
+		ins.Send(context.Background(), record.Record{Position: record.Position("test-pos")})
 	}
 }

pkg/inspector/inspector_test.go

@@ -29,7 +29,7 @@ import (
 
 func TestInspector_Send_NoSessions(t *testing.T) {
 	underTest := New(log.Nop(), 10)
-	underTest.Send(record.Record{})
+	underTest.Send(context.Background(), record.Record{})
 }
 
 func TestInspector_Send_SingleSession(t *testing.T) {
@@ -39,7 +39,7 @@ func TestInspector_Send_SingleSession(t *testing.T) {
 	r := record.Record{
 		Position: record.Position("test-pos"),
 	}
-	underTest.Send(r)
+	underTest.Send(context.Background(), r)
 	assertGotRecord(is.New(t), s, r)
 }
 
@@ -53,7 +53,7 @@ func TestInspector_Send_MultipleSessions(t *testing.T) {
 	r := record.Record{
 		Position: record.Position("test-pos"),
 	}
-	underTest.Send(r)
+	underTest.Send(context.Background(), r)
 	assertGotRecord(is, s1, r)
 	assertGotRecord(is, s2, r)
 }
@@ -67,12 +67,16 @@ func TestInspector_Send_SessionClosed(t *testing.T) {
 	r := record.Record{
 		Position: record.Position("test-pos"),
 	}
-	underTest.Send(r)
+	underTest.Send(context.Background(), r)
 	assertGotRecord(is.New(t), s, r)
 
-	err := cerrors.New("yet another error")
-	s.Close(err)
-	is.True(cerrors.Is(s.CloseReason, err))
+	s.Close()
+	underTest.Send(
+		context.Background(),
+		record.Record{
+			Position: record.Position("test-pos-2"),
+		},
+	)
 }
 
 func TestInspector_Send_SessionCtxCanceled(t *testing.T) {
@@ -85,12 +89,18 @@ func TestInspector_Send_SessionCtxCanceled(t *testing.T) {
 	r := record.Record{
 		Position: record.Position("test-pos"),
 	}
-	underTest.Send(r)
+	underTest.Send(context.Background(), r)
 	assertGotRecord(is.New(t), s, r)
 
 	cancel()
 	time.Sleep(100 * time.Millisecond)
-	is.True(cerrors.Is(s.CloseReason, context.Canceled))
+
+	select {
+	case _, ok := <-s.C:
+		is.True(!ok) // expected no record
+	default:
+		is.Fail() // expected channel to be closed
+	}
 }
 
 func TestInspector_Send_SlowConsumer(t *testing.T) {
@@ -106,9 +116,12 @@ func TestInspector_Send_SlowConsumer(t *testing.T) {
 	s := underTest.NewSession(context.Background())
 
 	for i := 0; i < bufferSize+1; i++ {
-		s.send(record.Record{
-			Position: record.Position(fmt.Sprintf("test-pos-%v", i)),
-		})
+		s.send(
+			context.Background(),
+			record.Record{
+				Position: record.Position(fmt.Sprintf("test-pos-%v", i)),
+			},
+		)
 	}
 
 	for i := 0; i < bufferSize; i++ {

pkg/orchestrator/connectors.go

@@ -16,26 +16,24 @@ package orchestrator
 
 import (
 	"context"
+
 	"github.com/conduitio/conduit/pkg/connector"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
 	"github.com/conduitio/conduit/pkg/foundation/rollback"
-	"github.com/conduitio/conduit/pkg/inspector"
 	"github.com/conduitio/conduit/pkg/pipeline"
+	"github.com/conduitio/conduit/pkg/record"
 	"github.com/google/uuid"
 )
 
 type ConnectorOrchestrator base
 
-func (c *ConnectorOrchestrator) Inspect(ctx context.Context, id string) (*inspector.Session, error) {
-	// fetch the connector from the ConnectorOrchestrator
+func (c *ConnectorOrchestrator) Inspect(ctx context.Context, id string) (chan record.Record, error) {
 	conn, err := c.Get(ctx, id)
 	if err != nil {
 		return nil, cerrors.Errorf("failed to get connector by ID %v: %w", id, err)
 	}
 
-	session := conn.Inspect(ctx)
-
-	return session, nil
+	return conn.Inspect(ctx).C, nil
 }
 
 func (c *ConnectorOrchestrator) Create(

pkg/web/api/connector_v1.go

@@ -19,9 +19,10 @@ package api
 import (
 	"context"
 	"fmt"
+
 	"github.com/conduitio/conduit/pkg/connector"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
-	"github.com/conduitio/conduit/pkg/inspector"
+	"github.com/conduitio/conduit/pkg/record"
 	"github.com/conduitio/conduit/pkg/web/api/fromproto"
 	"github.com/conduitio/conduit/pkg/web/api/status"
 	"github.com/conduitio/conduit/pkg/web/api/toproto"
@@ -36,7 +37,7 @@ type ConnectorOrchestrator interface {
 	Delete(ctx context.Context, id string) error
 	Update(ctx context.Context, id string, config connector.Config) (connector.Connector, error)
 	Validate(ctx context.Context, t connector.Type, config connector.Config) error
-	Inspect(ctx context.Context, id string) (*inspector.Session, error)
+	Inspect(ctx context.Context, id string) (chan record.Record, error)
 }
 
 type ConnectorAPIv1 struct {
@@ -73,26 +74,26 @@ func (c *ConnectorAPIv1) InspectConnector(req *apiv1.InspectConnectorRequest, se
 		return status.ConnectorError(cerrors.ErrEmptyID)
 	}
 
-	s, err := c.cs.Inspect(server.Context(), req.Id)
+	records, err := c.cs.Inspect(server.Context(), req.Id)
 	if err != nil {
 		return status.ConnectorError(cerrors.Errorf("failed to get connector by ID %v: %w", req.Id, err))
-
 	}
-	for r := range s.C {
-		record, err2 := toproto.Record(r)
+
+	for rec := range records {
+		recProto, err2 := toproto.Record(rec)
 		if err2 != nil {
 			return fmt.Errorf("failed converting record: %w", err2)
 		}
 
 		err2 = server.Send(&apiv1.InspectConnectorResponse{
-			Record: record,
+			Record: recProto,
 		})
 		if err2 != nil {
 			return fmt.Errorf("failed sending record: %w", err2)
 		}
 	}
 
-	return cerrors.Errorf("records channel closed: %w", s.CloseReason)
+	return cerrors.New("records channel closed")
 }
 
 // GetConnector returns a single Connector proto response or an error.

pkg/web/api/connector_v1_test.go

@@ -26,8 +26,6 @@ import (
 	"github.com/conduitio/conduit/pkg/foundation/assert"
 	"github.com/conduitio/conduit/pkg/foundation/cchan"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
-	"github.com/conduitio/conduit/pkg/foundation/log"
-	"github.com/conduitio/conduit/pkg/inspector"
 	"github.com/conduitio/conduit/pkg/record"
 	apimock "github.com/conduitio/conduit/pkg/web/api/mock"
 	"github.com/conduitio/conduit/pkg/web/api/toproto"
@@ -220,12 +218,11 @@ func TestConnectorAPIv1_InspectConnector_SendRecord(t *testing.T) {
 	recProto, err := toproto.Record(rec)
 	assert.Ok(t, err)
 
-	ins := inspector.New(log.Nop(), 10)
-	session := ins.NewSession(ctx)
+	records := make(chan record.Record)
 
 	csMock.EXPECT().
 		Inspect(ctx, id).
-		Return(session, nil).
+		Return(records, nil).
 		Times(1)
 
 	inspectServer := apimock.NewConnectorService_InspectConnectorServer(ctrl)
@@ -238,7 +235,7 @@ func TestConnectorAPIv1_InspectConnector_SendRecord(t *testing.T) {
 			inspectServer,
 		)
 	}()
-	ins.Send(rec)
+	records <- rec
 
 	time.Sleep(100 * time.Millisecond)
 }
@@ -250,12 +247,11 @@ func TestConnectorAPIv1_InspectConnector_SendErr(t *testing.T) {
 	csMock := apimock.NewConnectorOrchestrator(ctrl)
 	api := NewConnectorAPIv1(csMock)
 	id := uuid.NewString()
-	ins := inspector.New(log.Nop(), 10)
-	session := ins.NewSession(ctx)
+	records := make(chan record.Record)
 
 	csMock.EXPECT().
 		Inspect(ctx, id).
-		Return(session, nil).
+		Return(records, nil).
 		Times(1)
 
 	inspectServer := apimock.NewConnectorService_InspectConnectorServer(ctrl)
@@ -271,7 +267,7 @@ func TestConnectorAPIv1_InspectConnector_SendErr(t *testing.T) {
 		)
 		errC <- err
 	}()
-	ins.Send(generateTestRecord())
+	records <- generateTestRecord()
 
 	err, b, err2 := cchan.Chan[error](errC).RecvTimeout(context.Background(), 100*time.Millisecond)
 	assert.Ok(t, err2)