Fix grammatical errors and remove duplicate code in logs_router

This commit addresses several minor issues across the codebase:
- connector: Removed duplicate `Consumer` and `PipelineIDs` methods
  from `logsRouter` that shadowed the embedded `BaseRouter` methods.
  This also resolves a variable shadowing issue where `var errors error`
  shadowed the imported `errors` package.
- service/internal/graph: Fixed grammatically incorrect error messages
  in the graph builders that produced strings like "error creating
  exporter X is not supported".
- exporter/exporterhelper: Fixed grammar in queue comments ("consumer don't"
  to "consumers don't", and "allow distinguish" to "allow distinguishing").
- processor/batchprocessor: Fixed godoc grammar ("batcher is describes").

Author: farhan11166

connector/logs_router.go

@@ -4,11 +4,6 @@
 package connector // import "go.opentelemetry.io/collector/connector"
 
 import (
-	"errors"
-	"fmt"
-
-	"go.uber.org/multierr"
-
 	"go.opentelemetry.io/collector/connector/internal"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/internal/fanoutconsumer"
@@ -39,33 +34,4 @@ func NewLogsRouter(cm map[pipeline.ID]consumer.Logs) LogsRouterAndConsumer {
 	}
 }
 
-func (r *logsRouter) PipelineIDs() []pipeline.ID {
-	ids := make([]pipeline.ID, 0, len(r.Consumers))
-	for id := range r.Consumers {
-		ids = append(ids, id)
-	}
-	return ids
-}
-
-func (r *logsRouter) Consumer(pipelineIDs ...pipeline.ID) (consumer.Logs, error) {
-	if len(pipelineIDs) == 0 {
-		return nil, errors.New("missing consumers")
-	}
-	consumers := make([]consumer.Logs, 0, len(pipelineIDs))
-	var errors error
-	for _, pipelineID := range pipelineIDs {
-		c, ok := r.Consumers[pipelineID]
-		if ok {
-			consumers = append(consumers, c)
-		} else {
-			errors = multierr.Append(errors, fmt.Errorf("missing consumer: %q", pipelineID))
-		}
-	}
-	if errors != nil {
-		// TODO potentially this could return a NewLogs with the valid consumers
-		return nil, errors
-	}
-	return fanoutconsumer.NewLogs(consumers), nil
-}
-
 func (r *logsRouter) privateFunc() {}

exporter/exporterhelper/internal/queue/memory_queue.go

@@ -152,7 +152,7 @@ func (mq *memoryQueue[T]) Read(context.Context) (context.Context, T, Done, bool)
 			return context.Background(), el, nil, false
 		}
 
-		// TODO: Need to change the Queue interface to return an error to allow distinguish between shutdown and context canceled.
+		// TODO: Need to change the Queue interface to return an error to allow distinguishing between shutdown and context canceled.
 		//  Until then use the sync.Cond.
 		mq.hasMoreElements.Wait()
 	}

exporter/exporterhelper/internal/queue/persistent_queue.go

@@ -250,7 +250,7 @@ func (pq *persistentQueue[T]) Shutdown(ctx context.Context) error {
 
 	pq.mu.Lock()
 	defer pq.mu.Unlock()
-	// Mark this queue as stopped, so consumer don't start any more work.
+	// Mark this queue as stopped, so consumers don't start any more work.
 	pq.stopped = true
 	pq.hasMoreElements.Broadcast()
 	return pq.unrefClient(ctx)
@@ -345,7 +345,7 @@ func (pq *persistentQueue[T]) Read(ctx context.Context) (context.Context, T, Don
 			pq.hasMoreSpace.Signal()
 		}
 
-		// TODO: Need to change the Queue interface to return an error to allow distinguish between shutdown and context canceled.
+		// TODO: Need to change the Queue interface to return an error to allow distinguishing between shutdown and context canceled.
 		//  Until then use the sync.Cond.
 		pq.hasMoreElements.Wait()
 	}

processor/batchprocessor/batch_processor.go

@@ -57,7 +57,7 @@ type batchProcessor[T any] struct {
 	batcher batcher[T]
 }
 
-// batcher is describes a *singletonBatcher or *multiBatcher.
+// batcher describes a *singletonBatcher or *multiBatcher.
 type batcher[T any] interface {
 	// start initializes background resources used by this batcher.
 	start(ctx context.Context) error

service/internal/graph/exporter.go

@@ -98,7 +98,7 @@ func (n *exporterNode) buildComponent(
 		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), consumedSettings)
 		n.consumer = refconsumer.NewProfiles(n.consumer.(xconsumer.Profiles))
 	default:
-		return fmt.Errorf("error creating exporter %q for data type %q is not supported", set.ID, n.pipelineType)
+		return fmt.Errorf("failed to create exporter %q: data type %q is not supported", set.ID, n.pipelineType)
 	}
 	return nil
 }

service/internal/graph/processor.go

@@ -108,7 +108,7 @@ func (n *processorNode) buildComponent(ctx context.Context,
 		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), consumedSettings)
 		n.consumer = refconsumer.NewProfiles(n.consumer.(xconsumer.Profiles))
 	default:
-		return fmt.Errorf("error creating processor %q in pipeline %q, data type %q is not supported", set.ID, n.pipelineID.String(), n.pipelineID.Signal())
+		return fmt.Errorf("failed to create processor %q in pipeline %q: data type %q is not supported", set.ID, n.pipelineID.String(), n.pipelineID.Signal())
 	}
 	return nil
 }

service/internal/graph/receiver.go

@@ -92,7 +92,7 @@ func (n *receiverNode) buildComponent(ctx context.Context,
 		n.Component, err = builder.CreateProfiles(ctx, set,
 			obsconsumer.NewProfiles(fanoutconsumer.NewProfiles(consumers), producedSettings))
 	default:
-		return fmt.Errorf("error creating receiver %q for data type %q is not supported", set.ID, n.pipelineType)
+		return fmt.Errorf("failed to create receiver %q: data type %q is not supported", set.ID, n.pipelineType)
 	}
 	if err != nil {
 		return fmt.Errorf("failed to create %q receiver for data type %q: %w", set.ID, n.pipelineType, err)