[receiver/kafka] Ability to provide custom encoders

[xvilaca-te]

Component(s)

receiver/kafka

Is your feature request related to a problem? Please describe.

In our company, we use the kafkareceiver to consume messages from an internal Kafka instance, which are in a specific protobuf format. Up until now, we were able to use a custom encoder to deserialize those messages and convert them to the OTLP format. We achieved this by creating our own custom receiver where, in the factory function, we called the factory function with our own unmarshaller option:

kafkareceiver.NewFactory(kafkareceiver.WithMetricsUnmarshalers(customUnmarshaller))

where the customUnmarshaller implements the Unmarshaller interface and provides a name that we can use in the configuration yaml.

However, since this PR was merged, this is no longer supported. I understand that the change was performed to comply with the community requirements. We just want to understand what would be the best approach to unknot this tie.

Describe the solution you'd like

We noticed that it's still possible to pass options in the factory function and that the Unmarshaller interface is still being exported. Hence, an ideal solution for us would be to still allow to configure a custom unmarshaller via the options if possible. I'm just not sure how this could be done while still being compliant with checkapi.

Describe alternatives you've considered

We don't think that adding our custom unmarshaller to the set of supported unmarshallers is a good approach, since the data format is specific to our company.

One work-around for us is to use the raw encoder and then use a connector to convert the generated logs into metrics in the format we desire. However, we fear that this might incur additional performance penalties, and this solution will complicate our pipelines. A more direct solution would be ideal for us.

Additional context

No response

[github-actions]

Pinging code owners:

• receiver/kafka: @pavolloffay @MovieStoreGuy

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[antonjim-te]

cc @sakulali

[pavolloffay]

I am fine to expose the API for setting custom (un)marshaller.

[xvilaca-te]

I am fine to expose the API for setting custom (un)marshaller.

Thanks @pavolloffay Do we need the input of more people? And how could this be implemented without breaking the api checks that led you to make the breaking change in the first place?

[pavolloffay]

I am not sure why #27583 was implemented or if should be enforced. It seems like a limitation for people that want to extend components.

@mx-psi @dmitryax wdyt?

[mx-psi]

I don't think component Go modules are the right place to have public API other than the configuration and the factory.

We can expose Go API through a pkg module if it needs to live on contrib. Even then, it doesn't have to live there, it can just live on an external repository

[xvilaca-te]

I don't think component Go modules are the right place to have public API other than the configuration and the factory.

We can expose Go API through a pkg module if it needs to live on contrib. Even then, it doesn't have to live there, it can just live on an external repository

We are fine with referencing a separate repo if that's the best approach, as long as that repo is maintained. So, I guess that would mean you would have to keep the reference to contrib there up-to-date.

[mx-psi]

you would have to keep the reference to contrib there up-to-date.

It would just be one more dependency :) In any case, we can start with putting it under pkg, so long as we are fine with the stability guarantees

[xvilaca-te]

you would have to keep the reference to contrib there up-to-date.

It would just be one more dependency :) In any case, we can start with putting it under pkg, so long as we are fine with the stability guarantees

All good on our side. Just need to make sure that we can call the NewFactory method as above or something equivalent, i.e., we need the ability to create a kafka receiver factory extended with a list of custom unmarshallers.

Let me know what would be needed from our side. Thanks!

[atoulme]

Can we use the encoding approach we took for other components here? See #31774 for an example.

[kernelpanic77]

@atoulme I'd like to implement this. But I believe there is already an attribute for encoding in kafkareceiver/config.go.

[atoulme]

That is not the same thing. The current setting is a string. We want to allow to set a component.ID pointer to point to an extension.

[kernelpanic77]

@atoulme Correct me if I'm wrong but the use of the encoding approach would require the addition of the custom unmarshalling package extension/encoding, right ? (For this @xvilaca-te would again have to maintain a separate repository)

[wamphlett]

Is there any update on this? We rely on the WithTraceUnmarshallers function to handle custom message types and without this we are unable to upgrade our packages.

[xvilaca-te]

@kernelpanic77 @atoulme Any updates on this?

[xvilaca-te]

@kernelpanic77 @atoulme @mx-psi @pavolloffay any updates? Thanks!

[atoulme]

No updates. @kernelpanic77 please let us know how it's going

[xvilaca-te]

@atoulme @kernelpanic77 Thanks for the reply! We are now running into problems due to not being able to upgrade opentelemetry, so we would appreciate if this could be addressed as soon as possible. We are willing to contribute to it if it helps.