Skip to content

Allow build processors using kafka-clients Deserializer #267

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
kawamuray merged 5 commits into from
Sep 17, 2025
Merged

Allow build processors using kafka-clients Deserializer #267

kawamuray merged 5 commits into from
Sep 17, 2025

Conversation

@ocadaruma
Copy link
Member

@ocadaruma ocadaruma commented Sep 2, 2025 •
edited
Loading

Motivation

  • We sometimes want to use kafka-client's Deserializer directly on ProcessorsBuilder#consuming. for example:
    • To deserialize DecatonClient tasks using record-header information
  • Using TaskExtractor is not an option here, because:
    • When TaskExtractor is used, TaskMetadata construction becomes implementer's responsibility, while we intentionally avoid exposing public API to extract DecatonClient's taskMetadata from header

Summary of changes

  • ProcessorsBuilder#consuming now have an overload that accepts kafka.common.serialization.Deserializer<T>, which can be used to deserialize tasks using topic, record headers and custom TaskMetadata extraction logic is not necessary

@ocadaruma ocadaruma marked this pull request as ready for review September 2, 2025 06:02
@ocadaruma ocadaruma changed the title Allow customizing deserializer to apply transformation Allow build processors using kafka-clients Deserializer Sep 3, 2025
@ocadaruma ocadaruma requested a review from kawamuray September 3, 2025 14:21
@ocadaruma ocadaruma force-pushed the custom-deserializer branch 2 times, most recently from fa25a55 to 233b813 Compare September 8, 2025 13:08
kawamuray
kawamuray approved these changes Sep 17, 2025
View reviewed changes
processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorSubscription.java Outdated
consumeManager.close();
quotaApplier.close();
metrics.close();
processors.close();
Copy link
Contributor

@kawamuray kawamuray Sep 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hm, should we surround these close calls and log failure one-by-one?

Copy link
Member Author

@ocadaruma ocadaruma Sep 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, that might be good

kawamuray
kawamuray approved these changes Sep 17, 2025
View reviewed changes
Copy link
Contributor

@kawamuray kawamuray left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM, thanks

@kawamuray kawamuray merged commit e1ab6ad into line:master Sep 17, 2025
6 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@kawamuray kawamuray kawamuray approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@ocadaruma @kawamuray