resolved comments

agent/agent.go

@@ -30,7 +30,7 @@ type Agent struct {
 	retrier          *retrier
 	stopOnSinkError  bool
 	timerFn          TimerFn
-	batchSize        int
+	sinkBatchSize    int
 }
 
 // NewAgent returns an Agent with plugin factories.
@@ -52,7 +52,7 @@ func NewAgent(config Config) *Agent {
 		logger:           config.Logger,
 		retrier:          retrier,
 		timerFn:          timerFn,
-		batchSize:        config.BatchSize,
+		sinkBatchSize:    config.SinkBatchSize,
 	}
 }
 
@@ -313,7 +313,7 @@ func (r *Agent) setupSink(ctx context.Context, sr recipe.PluginRecipe, stream *s
 
 		r.logger.Info("Successfully published record", "sink", sr.Name, "recipe", recipeName)
 		return nil
-	}, r.batchSize)
+	}, r.sinkBatchSize)
 
 	stream.onClose(func() {
 		if err := sink.Close(); err != nil {

agent/agent_test.go

@@ -728,11 +728,6 @@ func TestAgentRun(t *testing.T) {
 		data := []models.Record{
 			models.NewRecord(&v1beta2.Asset{}),
 		}
-		batchedData := []models.Record{
-			models.NewRecord(&v1beta2.Asset{}),
-			models.NewRecord(&v1beta2.Asset{}),
-			models.NewRecord(&v1beta2.Asset{}),
-		}
 
 		extr := mocks.NewExtractor()
 		extr.SetEmit(data)
@@ -755,7 +750,7 @@ func TestAgentRun(t *testing.T) {
 
 		sink := mocks.NewSink()
 		sink.On("Init", mockCtx, buildPluginConfig(validRecipe.Sinks[0])).Return(nil).Once()
-		sink.On("Sink", mockCtx, batchedData).Return(nil)
+		sink.On("Sink", mockCtx, data).Return(nil)
 		sink.On("Close").Return(nil)
 		defer sink.AssertExpectations(t)
 
@@ -777,6 +772,7 @@ func TestAgentRun(t *testing.T) {
 			Monitor:              monitor,
 			MaxRetries:           2,                    // need to retry "at least" 2 times since Extractor returns RetryError twice
 			RetryInitialInterval: 1 * time.Millisecond, // this is to override default retry interval to reduce test time
+			SinkBatchSize:        1,
 		})
 		run := r.Run(ctx, validRecipe)
 		assert.NoError(t, run.Error)

agent/config.go

@@ -17,5 +17,5 @@ type Config struct {
 	RetryInitialInterval time.Duration
 	StopOnSinkError      bool
 	TimerFn              TimerFn
-	BatchSize            int
+	SinkBatchSize        int
 }

cmd/run.go

@@ -91,7 +91,7 @@ func RunCmd() *cobra.Command {
 				MaxRetries:           cfg.MaxRetries,
 				RetryInitialInterval: time.Duration(cfg.RetryInitialIntervalSeconds) * time.Second,
 				StopOnSinkError:      cfg.StopOnSinkError,
-				BatchSize:            cfg.BatchSize,
+				SinkBatchSize:        cfg.SinkBatchSize,
 			})
 
 			recipes, err := recipe.NewReader(lg, pathToConfig).Read(args[0])

config/config.go

@@ -17,7 +17,7 @@ type Config struct {
 	OtelEnabled                 bool    `mapstructure:"OTEL_ENABLED" default:"false"`
 	OtelCollectorAddr           string  `mapstructure:"OTEL_COLLECTOR_ADDR" default:"localhost:4317"`
 	OtelTraceSampleProbability  float64 `mapstructure:"OTEL_TRACE_SAMPLE_PROBABILITY" default:"1"`
-	BatchSize                   int     `mapstructure:"BATCH_SIZE" default:"1"`
+	SinkBatchSize               int     `mapstructure:"SINK_BATCH_SIZE" default:"1"`
 }
 
 func Load(configFile string) (Config, error) {

config/config_test.go

@@ -31,7 +31,7 @@ func TestLoad(t *testing.T) {
 				MaxRetries:                  5,
 				RetryInitialIntervalSeconds: 5,
 				StopOnSinkError:             false,
-				BatchSize:                   1,
+				SinkBatchSize:               1,
 			},
 		},
 		{
@@ -47,7 +47,7 @@ func TestLoad(t *testing.T) {
 				OtelTraceSampleProbability:  1,
 				MaxRetries:                  5,
 				RetryInitialIntervalSeconds: 5,
-				BatchSize:                   1,
+				SinkBatchSize:               1,
 			},
 			expectedErr: "",
 		},

config/meteor.yaml.sample

@@ -6,4 +6,4 @@ APP_NAME: meteor
 OTEL_ENABLED: false
 OTEL_COLLECTOR_ADDR: "localhost:4317"
 OTEL_TRACE_SAMPLE_PROBABILITY: 1
-BATCH_SIZE: 10
+SINK_BATCH_SIZE: 10