optional duckpaths channel

duck.go

@@ -248,7 +248,9 @@ func (o *Orchestrator[T]) DuckIngestWithRotate(ctx context.Context, w *sync.Wait
 			}
 			if o.duckConf.runner == nil || (o.duckConf.runner != nil && !o.duckConf.runner.IsDeleteDBOnDone()) {
 				o.duckConf.quack.Close()
-				o.duckPaths <- o.duckConf.quack.Path()
+				if o.opt.withDuckPathsChan {
+					o.duckPaths <- o.duckConf.quack.Path()
+				}
 			}
 			if !(o.rChanClosed && o.rChanRecs.Load() == 0) {
 				o.configureDuck(o.duckConf)

quacfka.go

@@ -27,6 +27,7 @@ type Opt struct {
 	withoutProc            bool
 	withoutDuck            bool
 	withoutDuckIngestRaw   bool
+	withDuckPathsChan      bool
 	fileRotateThresholdMB  int64
 	customArrow            []CustomArrow
 	normalizerFieldStrings []string
@@ -57,6 +58,12 @@ func WithoutDuck() Option {
 	}
 }
 
+func WithDuckPathsChan() Option {
+	return func(cfg config) {
+		cfg.withDuckPathsChan = true
+	}
+}
+
 func WithCustomArrows(p []CustomArrow) Option {
 	return func(cfg config) {
 		for _, c := range p {
@@ -151,7 +158,7 @@ func NewOrchestrator[T proto.Message](opts ...Option) (*Orchestrator[T], error)
 	o.rowGroupSizeMultiplier = 1
 	o.msgProcessorsCount.Store(1)
 	o.duckConnCount.Store(1)
-	o.duckPaths = make(chan string, 10)
+	o.duckPaths = make(chan string, 10000)
 	o.NewMetrics()
 	return o, nil
 }