Skip to content
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

Add config and header support for confluent schema registry. #10314

Merged
merged 14 commits into from
Feb 27, 2021
Merged

Add config and header support for confluent schema registry. #10314

merged 14 commits into from
Feb 27, 2021

Conversation

spinatelli
Copy link
Contributor

@spinatelli spinatelli commented Aug 24, 2020 •
edited
Loading

(porting code from #9096)

Fixes #8806.

Description

This is an update to the code in PR 9096 (#9096), since it's stale. I report the original description and the changes I made

Enhancing schema registry client i.e. SchemaRegistryBasedAvroBytesDecoder to accept additional configs, headers and able to query schemas from multi schema registry instances.

  • Changed SchemaRegistryBasedAvroBytesDecoder to accept config and headers as JSON properties.
  • Added support to pass multiple URLs for multiple schema registry instances by passing them as an array
  • Upgraded Confluent version to 5.5.1, which is compatible with existing kafka version: 2.6.0
  • Changed deprecated method getByID to getById.

Key changed/added classes in this PR

This PR has:

  • been self-reviewed.
  • added documentation for new or modified features or behaviors.
  • added or updated version, license, or notice information in licenses.yaml
  • added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
  • added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
  • been tested in a test Druid cluster.
Key changed/added classes in this PR
  • SchemaRegistryBasedAvroBytesDecoder
  • SchemaRegistryBasedAvroBytesDecoderTest
  • Updated libraries

@spinatelli spinatelli force-pushed the Druid-8806 branch 2 times, most recently from 06bf980 to 1b67a2f Compare August 24, 2020 17:57
@spinatelli spinatelli force-pushed the Druid-8806 branch 9 times, most recently from 26e49d0 to 765ed31 Compare August 26, 2020 08:18
@spinatelli spinatelli force-pushed the Druid-8806 branch 8 times, most recently from 7cd9813 to 72f5e4c Compare August 26, 2020 14:49
clintropolis
clintropolis reviewed Dec 2, 2020
View reviewed changes
Copy link
Member

@clintropolis clintropolis left a comment

Choose a reason for hiding this comment

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

👍

...ions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java Outdated
@@ -63,7 +73,7 @@ public GenericRecord parse(ByteBuffer bytes)
int id = bytes.getInt(); // extract schema registry id
int length = bytes.limit() - 1 - 4;
int offset = bytes.position() + bytes.arrayOffset();
Schema schema = registry.getByID(id);
Schema schema = registry.getById(id);
Copy link
Member

Choose a reason for hiding this comment

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

it seems like this method is deprecated as well, but it doesn't really seem like there is a better replacement since the underlying method is doing this

    ParsedSchema schema = this.getSchemaById(id);
    return schema instanceof AvroSchema ? ((AvroSchema)schema).rawSchema() : null;

Copy link
Contributor Author

@spinatelli spinatelli Feb 26, 2021 •
edited
Loading

Choose a reason for hiding this comment

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

Updated to use getSchemaById, which is anyway overridden by the CachedSchemaRegistryClient implementation that is being used here.

EDIT: actually just used that piece of code, it makes more sense I think, in case getById() gets changed in the future

licenses.yaml Outdated Show resolved Hide resolved
docs/ingestion/data-formats.md Outdated Show resolved Hide resolved
clintropolis
clintropolis approved these changes Feb 27, 2021
View reviewed changes
Copy link
Member

@clintropolis clintropolis 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 @spinatelli 👍

@clintropolis clintropolis merged commit 99198c0 into apache:master Feb 27, 2021
@clintropolis clintropolis mentioned this pull request Mar 1, 2021
Merged
8 tasks
@coldmav
Copy link

coldmav commented Mar 2, 2021

Hello all,
Any chance on adding the usage of a Password Provider to encapsulate the passwords, just like it is done on the Kafka Consumer properties (https://druid.apache.org/docs/latest/development/extensions-core/kafka-ingestion.html#kafkasupervisorioconfig) ?
Thank you all!

@spinatelli
Copy link
Contributor Author

Yea something like that would be the next step, I might look into it in the next week

@clintropolis clintropolis added this to the 0.22.0 milestone Aug 12, 2021
@clintropolis clintropolis mentioned this pull request Sep 3, 2021
Closed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@clintropolis clintropolis clintropolis approved these changes

Assignees
No one assigned
Labels
Area - Extension Area - Streaming Ingestion
Projects
None yet
Milestone
0.22.0
Development

Successfully merging this pull request may close these issues.

Support Confluent Schema Registry With Authentication
3 participants
@spinatelli @coldmav @clintropolis