Change With*Unmarshallers signatures (#2973)

* Change With*Unmarshallers signatures; Rename variables in kafkaReceiverFactory

* fix tests

* rollback change to variables in kafkaReceiverFactory

exporter/kafkaexporter/factory.go

@@ -41,11 +41,11 @@ const (
 // FactoryOption applies changes to kafkaExporterFactory.
 type FactoryOption func(factory *kafkaExporterFactory)
 
-// WithAddTracesMarshallers adds tracesMarshallers.
-func WithAddTracesMarshallers(encodingMarshaller map[string]TracesMarshaller) FactoryOption {
+// WithTracesMarshallers adds tracesMarshallers.
+func WithTracesMarshallers(tracesMarshallers ...TracesMarshaller) FactoryOption {
 	return func(factory *kafkaExporterFactory) {
-		for encoding, marshaller := range encodingMarshaller {
-			factory.tracesMarshallers[encoding] = marshaller
+		for _, marshaller := range tracesMarshallers {
+			factory.tracesMarshallers[marshaller.Encoding()] = marshaller
 		}
 	}
 }

exporter/kafkaexporter/factory_test.go

@@ -104,7 +104,7 @@ func TestCreateLogsExporter_err(t *testing.T) {
 
 func TestWithMarshallers(t *testing.T) {
 	cm := &customMarshaller{}
-	f := NewFactory(WithAddTracesMarshallers(map[string]TracesMarshaller{cm.Encoding(): cm}))
+	f := NewFactory(WithTracesMarshallers(cm))
 	cfg := createDefaultConfig().(*Config)
 	// disable contacting broker
 	cfg.Metadata.Full = false

receiver/kafkareceiver/factory.go

@@ -44,20 +44,20 @@ const (
 // FactoryOption applies changes to kafkaReceiverFactory.
 type FactoryOption func(factory *kafkaReceiverFactory)
 
-// WithAddTracesUnmarshallers adds marshallers.
-func WithAddTracesUnmarshallers(encodingMarshaller map[string]TracesUnmarshaller) FactoryOption {
+// WithTracesUnmarshallers adds Unmarshallers.
+func WithTracesUnmarshallers(tracesUnmarshallers ...TracesUnmarshaller) FactoryOption {
 	return func(factory *kafkaReceiverFactory) {
-		for encoding, unmarshaller := range encodingMarshaller {
-			factory.tracesUnmarshalers[encoding] = unmarshaller
+		for _, unmarshaller := range tracesUnmarshallers {
+			factory.tracesUnmarshalers[unmarshaller.Encoding()] = unmarshaller
 		}
 	}
 }
 
-// WithAddLogsUnmarshallers adds LogsUnmarshallers.
-func WithAddLogsUnmarshallers(encodingMarshaller map[string]LogsUnmarshaller) FactoryOption {
+// WithLogsUnmarshallers adds LogsUnmarshallers.
+func WithLogsUnmarshallers(logsUnmarshallers ...LogsUnmarshaller) FactoryOption {
 	return func(factory *kafkaReceiverFactory) {
-		for encoding, unmarshaller := range encodingMarshaller {
-			factory.logsUnmarshaller[encoding] = unmarshaller
+		for _, unmarshaller := range logsUnmarshallers {
+			factory.logsUnmarshaller[unmarshaller.Encoding()] = unmarshaller
 		}
 	}
 }

receiver/kafkareceiver/factory_test.go

@@ -60,7 +60,7 @@ func TestCreateTracesReceiver_error(t *testing.T) {
 
 func TestWithTracesUnmarshallers(t *testing.T) {
 	unmarshaller := &customTracesUnmarshaller{}
-	f := NewFactory(WithAddTracesUnmarshallers(map[string]TracesUnmarshaller{unmarshaller.Encoding(): unmarshaller}))
+	f := NewFactory(WithTracesUnmarshallers(unmarshaller))
 	cfg := createDefaultConfig().(*Config)
 	// disable contacting broker
 	cfg.Metadata.Full = false
@@ -104,7 +104,7 @@ func TestCreateLogsReceiver_error(t *testing.T) {
 
 func TestWithLogsUnmarshallers(t *testing.T) {
 	unmarshaller := &customLogsUnmarshaller{}
-	f := NewFactory(WithAddLogsUnmarshallers(map[string]LogsUnmarshaller{unmarshaller.Encoding(): unmarshaller}))
+	f := NewFactory(WithLogsUnmarshallers(unmarshaller))
 	cfg := createDefaultConfig().(*Config)
 	// disable contacting broker
 	cfg.Metadata.Full = false