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.

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

MAINT: allow latest schema version if not specified in confluent schema #4453

Merged

Conversation

chenqi0805
Copy link
Collaborator

@chenqi0805 chenqi0805 commented Apr 23, 2024

Description

This PR fixes the requirement to specify schema version in confluent schema type:

current

kafka:
      topics:
        - name: "topic_4"
          group_id: "groupdID1"
      bootstrap_servers:
        - ...
      schema:
        type: confluent
        registry_url: https://psrc-m5k9x.us-west-2.aws.confluent.cloud
        version: 1 # <- required previously

after this change:

kafka:
      topics:
        - name: "topic_4"
          group_id: "groupdID1"
      bootstrap_servers:
        - ...
      schema:
        type: confluent
        registry_url: https://psrc-m5k9x.us-west-2.aws.confluent.cloud

Run confluent integration tests manually to verify the change

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

  • New functionality includes testing.
  • New functionality has a documentation issue. Please link to it in this PR.
    • New functionality has javadoc added
  • Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

@@ -293,9 +293,15 @@ private void setPropertiesForSchemaType(Properties properties, TopicConfig topic
properties.put("auto.register.schemas", false);
schemaRegistryClient = new CachedSchemaRegistryClient(getSchemaRegistryUrl(),
100, propertyMap);
final SchemaConfig schemaConfig = sourceConfig.getSchemaConfig();
Copy link
Member

Choose a reason for hiding this comment

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

Can we combine this code with the code above?

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

I believe it has been attempted by myself and others to use KafkaCustomConsumerFactory here but there has been some subtleties. Tracked in a separate issue: #4456

@chenqi0805 chenqi0805 merged commit c31aa6a into opensearch-project:main Apr 24, 2024
44 of 50 checks passed
@chenqi0805 chenqi0805 deleted the fix/confluent-schema-version branch April 24, 2024 16:13
srikanthjg pushed a commit to srikanthjg/data-prepper that referenced this pull request Apr 24, 2024
@kkondaka kkondaka added this to the v2.8 milestone May 14, 2024
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

4 participants