[9.4](backport #50341) Revert "fix(otelconsumer): do not retry 401 errors from Elasticsearch" (#50353)

* Revert "fix(otelconsumer): do not retry 401 errors from Elasticsearch (#50261)" (#50341)

This reverts commit fdea8cd.

(cherry picked from commit 4ed1c3a)

# Conflicts:
#	libbeat/otel/otelconsumer/otelconsumer.go

* Fix conflict

---------

Co-authored-by: Andrzej Stencel <andrzej.stencel@elastic.co>
Co-authored-by: Craig MacKenzie <craig.mackenzie@elastic.co>

Author: 3 people

libbeat/otel/otelconsumer/otelconsumer.go

@@ -21,7 +21,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"net/http"
 	"os"
 	"runtime"
 	"time"
@@ -54,12 +53,6 @@ func init() {
 	outputs.RegisterType("otelconsumer", makeOtelConsumer)
 }
 
-// statusCodeError is satisfied by errors that carry an HTTP status code,
-// such as docappender.ErrorFlushFailed errors returned from the OTelCol Elasticsearch exporter.
-type statusCodeError interface {
-	StatusCode() int
-}
-
 type otelConsumer struct {
 	observer       outputs.Observer
 	logsConsumer   consumer.Logs
@@ -211,20 +204,13 @@ func (out *otelConsumer) logsPublish(ctx context.Context, batch publisher.Batch)
 
 	err := out.logsConsumer.ConsumeLogs(otelctx.NewConsumerContext(ctx, out.beatInfo), pLogs)
 	if err != nil {
-		// Work around the fact that Elasticsearch exporter returns 401 as a non-permanent error.
-		isAuthorizationError := false
-		var statusErr statusCodeError
-		if errors.As(err, &statusErr) {
-			isAuthorizationError = statusErr.StatusCode() == http.StatusUnauthorized
-		}
-
 		// Permanent errors shouldn't be retried. This tipically means
 		// the data cannot be serialized by the exporter that is attached
 		// to the pipeline or when the destination refuses the data because
 		// it cannot decode it. Retrying in this case is useless.
 		//
 		// See https://github.com/open-telemetry/opentelemetry-collector/blob/1c47d89/receiver/doc.go#L23-L40
-		if consumererror.IsPermanent(err) || isAuthorizationError {
+		if consumererror.IsPermanent(err) {
 			st.PermanentErrors(len(events))
 			batch.Drop()
 		} else {

libbeat/otel/otelconsumer/otelconsumer_test.go

@@ -266,32 +266,6 @@ func TestPublish(t *testing.T) {
 		assert.Equal(t, outest.BatchRetry, batch.Signals[0].Tag)
 	})
 
-	t.Run("drops batch on 401 Unauthorized error", func(t *testing.T) {
-		batch := outest.NewBatch(event1, event2, event3)
-
-		otelConsumer := makeOtelConsumer(t, func(ctx context.Context, ld plog.Logs) error {
-			return &testStatusCodeError{statusCode: 401, msg: "flush failed (401): unauthorized"}
-		})
-
-		err := otelConsumer.Publish(ctx, batch)
-		assert.NoError(t, err)
-		assert.Len(t, batch.Signals, 1)
-		assert.Equal(t, outest.BatchDrop, batch.Signals[0].Tag)
-	})
-
-	t.Run("retries batch on non-401 status code error", func(t *testing.T) {
-		batch := outest.NewBatch(event1, event2, event3)
-
-		otelConsumer := makeOtelConsumer(t, func(ctx context.Context, ld plog.Logs) error {
-			return &testStatusCodeError{statusCode: 500, msg: "flush failed (500): internal server error"}
-		})
-
-		err := otelConsumer.Publish(ctx, batch)
-		assert.NoError(t, err)
-		assert.Len(t, batch.Signals, 1)
-		assert.Equal(t, outest.BatchRetry, batch.Signals[0].Tag)
-	})
-
 	t.Run("sets the elasticsearchexporter doc id attribute from metadata", func(t *testing.T) {
 		batch := outest.NewBatch(event4)
 
@@ -416,11 +390,3 @@ func checkEventsActive(reg *monitoring.Registry) int64 {
 	outputSnapshot := monitoring.CollectFlatSnapshot(reg, monitoring.Full, true)
 	return outputSnapshot.Ints["events.active"]
 }
-
-type testStatusCodeError struct {
-	statusCode int
-	msg        string
-}
-
-func (e *testStatusCodeError) Error() string   { return e.msg }
-func (e *testStatusCodeError) StatusCode() int { return e.statusCode }