receiver/kafkareceiver: fix hyphenated text encodings (open-telemetry#39806)

#### Description

Fix support for text encodings with hyphens in their names.

If the encoding name has a hyphen then it is an invalid extension ID,
but we should not return an error due to this if it's a built-in
encoding.

#### Link to tracking issue

Fixes open-telemetry#39793

#### Testing

Added a new unit test covering hyphenated text encoding names (fails
without the associated fix).

#### Documentation

N/A

Author: axw

.chloggen/kafkareceiver-textencoding.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix support for built-in text encodings with hyphens in the encoding name
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39793]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/kafkareceiver/encoding.go

@@ -20,12 +20,15 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver/internal/unmarshaler"
 )
 
-var errUnknownEncodingExtension = errors.New("unknown encoding extension")
+var (
+	errUnknownEncodingExtension = errors.New("unknown encoding extension")
+	errInvalidComponentType     = errors.New("invalid component type")
+)
 
 func newTracesUnmarshaler(encoding string, _ receiver.Settings, host component.Host) (ptrace.Unmarshaler, error) {
 	// Extensions take precedence.
 	if unmarshaler, err := loadEncodingExtension[ptrace.Unmarshaler](host, encoding, "traces"); err != nil {
-		if !errors.Is(err, errUnknownEncodingExtension) {
+		if !errors.Is(err, errInvalidComponentType) && !errors.Is(err, errUnknownEncodingExtension) {
 			return nil, err
 		}
 	} else {
@@ -53,7 +56,7 @@ func newTracesUnmarshaler(encoding string, _ receiver.Settings, host component.H
 func newLogsUnmarshaler(encoding string, set receiver.Settings, host component.Host) (plog.Unmarshaler, error) {
 	// Extensions take precedence.
 	if unmarshaler, err := loadEncodingExtension[plog.Unmarshaler](host, encoding, "logs"); err != nil {
-		if !errors.Is(err, errUnknownEncodingExtension) {
+		if !errors.Is(err, errInvalidComponentType) && !errors.Is(err, errUnknownEncodingExtension) {
 			return nil, err
 		}
 	} else {
@@ -91,7 +94,7 @@ func newLogsUnmarshaler(encoding string, set receiver.Settings, host component.H
 func newMetricsUnmarshaler(encoding string, _ receiver.Settings, host component.Host) (pmetric.Unmarshaler, error) {
 	// Extensions take precedence.
 	if unmarshaler, err := loadEncodingExtension[pmetric.Unmarshaler](host, encoding, "metrics"); err != nil {
-		if !errors.Is(err, errUnknownEncodingExtension) {
+		if !errors.Is(err, errInvalidComponentType) && !errors.Is(err, errUnknownEncodingExtension) {
 			return nil, err
 		}
 	} else {
@@ -128,7 +131,7 @@ func loadEncodingExtension[T any](host component.Host, encoding, signalType stri
 func encodingToComponentID(encoding string) (*component.ID, error) {
 	componentType, err := component.NewType(encoding)
 	if err != nil {
-		return nil, fmt.Errorf("invalid component type: %w", err)
+		return nil, fmt.Errorf("%w: %w", errInvalidComponentType, err)
 	}
 	id := component.NewID(componentType)
 	return &id, nil

receiver/kafkareceiver/encoding_test.go

@@ -131,6 +131,19 @@ func TestNewLogsUnmarshaler(t *testing.T) {
 				assert.NoError(t, plogtest.CompareLogs(logs, actual, plogtest.IgnoreObservedTimestamp()))
 			},
 		},
+		{
+			encoding: "text_utf-16", // hyphen in the name
+			input: func() []byte {
+				out, _ := unicode.UTF16(
+					unicode.LittleEndian,
+					unicode.IgnoreBOM,
+				).NewEncoder().Bytes([]byte("hello world"))
+				return out
+			}(),
+			check: func(t *testing.T, actual plog.Logs) {
+				assert.NoError(t, plogtest.CompareLogs(logs, actual, plogtest.IgnoreObservedTimestamp()))
+			},
+		},
 	} {
 		t.Run(tc.encoding, func(t *testing.T) {
 			u := mustNewLogsUnmarshaler(t, tc.encoding, componenttest.NewNopHost())