[
https://issues.apache.org/jira/browse/CAMEL-24172?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Carles Arnal updated CAMEL-24172:
---------------------------------
Description:
I'd like to propose a new component for integrating with *Apicurio Registry.*
Camel has Confluent Schema Registry support, but nothing for Apicurio, which is
widely used in the Kafka ecosystem. Users currently have to write custom beans
or raw HTTP calls to interact with it from Camel routes.
The idea is a camel-apicurio-registry component built on the official Java SDK
(available in Maven Central) that would support:
- *Producer ops:* create/update/delete artifacts, get content/metadata,
search, test compatibility
- *Consumer:* poll for new versions or consume webhook notifications on
artifact lifecycle events
apicurio-registry:my-group/my-schema?registryUrl=[http://localhost:8080/apis/registry/v3]
I'm opening this based on the conversation in [#camel > New component proposal:
camel-apicurio-registry @
💬|https://camel.zulipchat.com/#narrow/channel/257298-camel/topic/New.20component.20proposal.3A.20camel-apicurio-registry/near/611041432].
h2. *Use Cases*
{*}Use Case 1: Schema Governance in Kafka Pipelines{*}{*}{{*}}
Register and validate schemas before producing Kafka messages. Ensures all
producers use a centrally governed schema, preventing schema drift across teams.
{code:java}
// Register an Avro schema before starting to produce
from("file:schemas?include=.*\\.avsc")
.setHeader("CamelApicurioRegistryOperation", constant("createArtifact"))
.setHeader("CamelApicurioRegistryGroupId", constant("kafka-schemas"))
.setHeader("CamelApicurioRegistryArtifactType", constant("AVRO"))
.setHeader("CamelApicurioRegistryIfExists", constant("CREATE_VERSION"))
.to("apicurio-registry://kafka-schemas?registryUrl={{registry.url}}")
.log("Registered schema: ${header.CamelApicurioRegistryArtifactId}");
{code}
{*}Use Case 2: Contract-First API Development{*}{*}{{*}}
Pull an OpenAPI spec from the registry and use it to validate incoming HTTP
requests, ensuring API implementations stay in sync with the contract.
{code:java}
// Fetch the latest OpenAPI spec from the registry and validate requests
from("timer:refreshSpec?period=60000")
.setHeader("CamelApicurioRegistryOperation",
constant("getArtifactContent"))
.to("apicurio-registry://my-apis/orders-api?registryUrl={{registry.url}}")
.convertBodyTo(String.class)
.to("direct:updateValidator");
from("platform-http:/api/orders")
.to("direct:validateAgainstSpec")
.to("bean:orderService");
{code}
{*}Use Case 3: Schema Evolution Notifications{*}{*}{{*}}
Poll the registry for new artifact versions and notify downstream systems
(Slack, email, CI/CD pipelines) when schemas change — essential for data mesh
architectures.
{code:java}
// React to new schema versions
from("apicurio-registry://my-group/payment-event?registryUrl={{registry.url}}&delay=30000")
.log("New version of payment-event schema:
v${header.CamelApicurioRegistryVersion}")
.choice()
.when(header("CamelApicurioRegistryArtifactType").isEqualTo("AVRO"))
.to("slack:#schema-changes")
.when(header("CamelApicurioRegistryArtifactType").isEqualTo("OPENAPI"))
.to("webhook:https://ci.example.com/rebuild-sdk";)
.end();
{code}
{*}Use Case 4: Multi-Format Schema Migration{*}{*}{{*}}
Migrate schemas across environments (dev → staging → prod) or convert between
formats. Useful for organizations managing schemas across multiple clusters.
{code:java}
// Export all schemas from one registry and import into another
from("apicurio-registry://my-group?registryUrl={{registry.dev.url}}&operation=searchArtifacts")
.split(body())
.setHeader("CamelApicurioRegistryOperation",
constant("getArtifactContent"))
.toD("apicurio-registry://${header.CamelApicurioRegistryGroupId}/${header.CamelApicurioRegistryArtifactId}?registryUrl={{registry.dev.url}}")
.setHeader("CamelApicurioRegistryOperation",
constant("createArtifact"))
.setHeader("CamelApicurioRegistryIfExists",
constant("CREATE_VERSION"))
.to("apicurio-registry://my-group?registryUrl={{registry.prod.url}}")
.end()
.log("Schema migration complete");
{code}
{*}Use Case 5: AsyncAPI-Driven Event Router{*}{*}{{*}}
Use AsyncAPI specs stored in the registry to dynamically configure Camel routes
for event-driven architectures — the registry becomes the source of truth for
messaging topology.
{code:java}
// Read an AsyncAPI spec and route events to the correct channel
from("timer:configRefresh?period=120000")
.setHeader("CamelApicurioRegistryOperation",
constant("getArtifactContent"))
.to("apicurio-registry://event-platform/order-events?registryUrl={{registry.url}}&artifactType=ASYNCAPI")
.bean("asyncApiRouteBuilder", "updateRoutes");
{code}
{*}Use Case 6: Schema Compatibility Gate in CI/CD{*}{*}{{*}}
Integrate into a Tekton/Jenkins pipeline to test schema compatibility before
merging changes. Fail the build if a new schema version would break consumers.
{code:java}
// CI/CD compatibility check
from("direct:checkCompatibility")
.setHeader("CamelApicurioRegistryOperation",
constant("testCompatibility"))
.to("apicurio-registry://my-group/user-event?registryUrl={{registry.url}}")
.choice()
.when(header("CamelApicurioRegistryCompatible").isEqualTo(true))
.log("Schema is compatible — safe to deploy")
.to("direct:proceed")
.otherwise()
.log("BREAKING CHANGE detected!")
.to("direct:failBuild")
.end();
{code}
{*}Use Case 7: Data Format — Schema-Validated Marshalling{*}{*}{{*}}
Use Apicurio Registry as a Camel DataFormat for serialization/deserialization
in any route (not just Kafka), pulling schemas from the registry at runtime.
{code:java}
// Unmarshal JMS messages using Avro schema from the registry
from("jms:queue:orders")
.unmarshal().custom("apicurio-avro:my-group/order-schema")
.process(exchange -> {
GenericRecord order = exchange.getIn().getBody(GenericRecord.class);
// process the order...
})
.marshal().custom("apicurio-avro:my-group/order-response-schema")
.to("jms:queue:order-responses");
{code}
{*}Supported Artifact Types{*}{*}{{*}}
Apicurio Registry supports a wide range of artifact types beyond traditional
schemas, making the component useful across many integration scenarios.
was:
I'd like to propose a new component for integrating with *Apicurio Registry.*
Camel has Confluent Schema Registry support, but nothing for Apicurio, which is
widely used in the Kafka ecosystem. Users currently have to write custom beans
or raw HTTP calls to interact with it from Camel routes.
The idea is a camel-apicurio-registry component built on the official Java SDK
(available in Maven Central) that would support:
- *Producer ops:* create/update/delete artifacts, get content/metadata,
search, test compatibility
- *Consumer:* poll for new versions or consume webhook notifications on
artifact lifecycle events
apicurio-registry:my-group/my-schema?registryUrl=[http://localhost:8080/apis/registry/v3]
I'm opening this based on the conversation in [#camel > New component proposal:
camel-apicurio-registry @
💬|https://camel.zulipchat.com/#narrow/channel/257298-camel/topic/New.20component.20proposal.3A.20camel-apicurio-registry/near/611041432].
{*}Use Cases and Camel Route Examples{*}{*}{*}
{*}Use Case 1: Schema Governance in Kafka Pipelines{*}{*}{*}
Register and validate schemas before producing Kafka messages. Ensures all
producers use a centrally governed schema, preventing schema drift across teams.
{code:java}
// Register an Avro schema before starting to produce
from("file:schemas?include=.*\\.avsc")
.setHeader("CamelApicurioRegistryOperation", constant("createArtifact"))
.setHeader("CamelApicurioRegistryGroupId", constant("kafka-schemas"))
.setHeader("CamelApicurioRegistryArtifactType", constant("AVRO"))
.setHeader("CamelApicurioRegistryIfExists", constant("CREATE_VERSION"))
.to("apicurio-registry://kafka-schemas?registryUrl={{registry.url}}")
.log("Registered schema: ${header.CamelApicurioRegistryArtifactId}");
{code}
{*}Use Case 2: Contract-First API Development{*}{*}{*}
Pull an OpenAPI spec from the registry and use it to validate incoming HTTP
requests, ensuring API implementations stay in sync with the contract.
{code:java}
// Fetch the latest OpenAPI spec from the registry and validate requests
from("timer:refreshSpec?period=60000")
.setHeader("CamelApicurioRegistryOperation",
constant("getArtifactContent"))
.to("apicurio-registry://my-apis/orders-api?registryUrl={{registry.url}}")
.convertBodyTo(String.class)
.to("direct:updateValidator");
from("platform-http:/api/orders")
.to("direct:validateAgainstSpec")
.to("bean:orderService");
{code}
{*}Use Case 3: Schema Evolution Notifications{*}{*}{*}
Poll the registry for new artifact versions and notify downstream systems
(Slack, email, CI/CD pipelines) when schemas change — essential for data mesh
architectures.
{code:java}
// React to new schema versions
from("apicurio-registry://my-group/payment-event?registryUrl={{registry.url}}&delay=30000")
.log("New version of payment-event schema:
v${header.CamelApicurioRegistryVersion}")
.choice()
.when(header("CamelApicurioRegistryArtifactType").isEqualTo("AVRO"))
.to("slack:#schema-changes")
.when(header("CamelApicurioRegistryArtifactType").isEqualTo("OPENAPI"))
.to("webhook:https://ci.example.com/rebuild-sdk";)
.end();
{code}
{*}Use Case 4: Multi-Format Schema Migration{*}{*}{*}
Migrate schemas across environments (dev → staging → prod) or convert between
formats. Useful for organizations managing schemas across multiple clusters.
{code:java}
// Export all schemas from one registry and import into another
from("apicurio-registry://my-group?registryUrl={{registry.dev.url}}&operation=searchArtifacts")
.split(body())
.setHeader("CamelApicurioRegistryOperation",
constant("getArtifactContent"))
.toD("apicurio-registry://${header.CamelApicurioRegistryGroupId}/${header.CamelApicurioRegistryArtifactId}?registryUrl={{registry.dev.url}}")
.setHeader("CamelApicurioRegistryOperation",
constant("createArtifact"))
.setHeader("CamelApicurioRegistryIfExists",
constant("CREATE_VERSION"))
.to("apicurio-registry://my-group?registryUrl={{registry.prod.url}}")
.end()
.log("Schema migration complete");
{code}
{*}Use Case 5: AsyncAPI-Driven Event Router{*}{*}{*}
Use AsyncAPI specs stored in the registry to dynamically configure Camel routes
for event-driven architectures — the registry becomes the source of truth for
messaging topology.
{code:java}
// Read an AsyncAPI spec and route events to the correct channel
from("timer:configRefresh?period=120000")
.setHeader("CamelApicurioRegistryOperation",
constant("getArtifactContent"))
.to("apicurio-registry://event-platform/order-events?registryUrl={{registry.url}}&artifactType=ASYNCAPI")
.bean("asyncApiRouteBuilder", "updateRoutes");
{code}
{*}Use Case 6: Schema Compatibility Gate in CI/CD{*}{*}{*}
Integrate into a Tekton/Jenkins pipeline to test schema compatibility before
merging changes. Fail the build if a new schema version would break consumers.
{code:java}
// CI/CD compatibility check
from("direct:checkCompatibility")
.setHeader("CamelApicurioRegistryOperation",
constant("testCompatibility"))
.to("apicurio-registry://my-group/user-event?registryUrl={{registry.url}}")
.choice()
.when(header("CamelApicurioRegistryCompatible").isEqualTo(true))
.log("Schema is compatible — safe to deploy")
.to("direct:proceed")
.otherwise()
.log("BREAKING CHANGE detected!")
.to("direct:failBuild")
.end();
{code}
{*}Use Case 7: Data Format — Schema-Validated Marshalling{*}{*}{*}
Use Apicurio Registry as a Camel DataFormat for serialization/deserialization
in any route (not just Kafka), pulling schemas from the registry at runtime.
{code:java}
// Unmarshal JMS messages using Avro schema from the registry
from("jms:queue:orders")
.unmarshal().custom("apicurio-avro:my-group/order-schema")
.process(exchange -> {
GenericRecord order = exchange.getIn().getBody(GenericRecord.class);
// process the order...
})
.marshal().custom("apicurio-avro:my-group/order-response-schema")
.to("jms:queue:order-responses");
{code}
{*}Supported Artifact Types{*}{*}{*}
Apicurio Registry supports a wide range of artifact types beyond traditional
schemas, making the component useful across many integration scenarios.
> New component proposal: camel-apicurio-registry
> -----------------------------------------------
>
> Key: CAMEL-24172
> URL: https://issues.apache.org/jira/browse/CAMEL-24172
> Project: Camel
> Issue Type: New Feature
> Reporter: Carles Arnal
> Priority: Major
>
> I'd like to propose a new component for integrating with *Apicurio Registry.*
> Camel has Confluent Schema Registry support, but nothing for Apicurio, which
> is widely used in the Kafka ecosystem. Users currently have to write custom
> beans or raw HTTP calls to interact with it from Camel routes.
> The idea is a camel-apicurio-registry component built on the official Java
> SDK (available in Maven Central) that would support:
> - *Producer ops:* create/update/delete artifacts, get content/metadata,
> search, test compatibility
> - *Consumer:* poll for new versions or consume webhook notifications on
> artifact lifecycle events
>
> apicurio-registry:my-group/my-schema?registryUrl=[http://localhost:8080/apis/registry/v3]
> I'm opening this based on the conversation in [#camel > New component
> proposal: camel-apicurio-registry @
> 💬|https://camel.zulipchat.com/#narrow/channel/257298-camel/topic/New.20component.20proposal.3A.20camel-apicurio-registry/near/611041432].
> h2. *Use Cases*
> {*}Use Case 1: Schema Governance in Kafka Pipelines{*}{*}{{*}}
> Register and validate schemas before producing Kafka messages. Ensures all
> producers use a centrally governed schema, preventing schema drift across
> teams.
> {code:java}
> // Register an Avro schema before starting to produce
> from("file:schemas?include=.*\\.avsc")
> .setHeader("CamelApicurioRegistryOperation", constant("createArtifact"))
> .setHeader("CamelApicurioRegistryGroupId", constant("kafka-schemas"))
> .setHeader("CamelApicurioRegistryArtifactType", constant("AVRO"))
> .setHeader("CamelApicurioRegistryIfExists", constant("CREATE_VERSION"))
> .to("apicurio-registry://kafka-schemas?registryUrl={{registry.url}}")
> .log("Registered schema: ${header.CamelApicurioRegistryArtifactId}");
> {code}
> {*}Use Case 2: Contract-First API Development{*}{*}{{*}}
> Pull an OpenAPI spec from the registry and use it to validate incoming HTTP
> requests, ensuring API implementations stay in sync with the contract.
> {code:java}
> // Fetch the latest OpenAPI spec from the registry and validate requests
> from("timer:refreshSpec?period=60000")
> .setHeader("CamelApicurioRegistryOperation",
> constant("getArtifactContent"))
>
> .to("apicurio-registry://my-apis/orders-api?registryUrl={{registry.url}}")
> .convertBodyTo(String.class)
> .to("direct:updateValidator");
> from("platform-http:/api/orders")
> .to("direct:validateAgainstSpec")
> .to("bean:orderService");
> {code}
> {*}Use Case 3: Schema Evolution Notifications{*}{*}{{*}}
> Poll the registry for new artifact versions and notify downstream systems
> (Slack, email, CI/CD pipelines) when schemas change — essential for data mesh
> architectures.
> {code:java}
> // React to new schema versions
>
> from("apicurio-registry://my-group/payment-event?registryUrl={{registry.url}}&delay=30000")
> .log("New version of payment-event schema:
> v${header.CamelApicurioRegistryVersion}")
> .choice()
> .when(header("CamelApicurioRegistryArtifactType").isEqualTo("AVRO"))
> .to("slack:#schema-changes")
> .when(header("CamelApicurioRegistryArtifactType").isEqualTo("OPENAPI"))
> .to("webhook:https://ci.example.com/rebuild-sdk";)
> .end();
> {code}
> {*}Use Case 4: Multi-Format Schema Migration{*}{*}{{*}}
> Migrate schemas across environments (dev → staging → prod) or convert between
> formats. Useful for organizations managing schemas across multiple clusters.
> {code:java}
> // Export all schemas from one registry and import into another
>
> from("apicurio-registry://my-group?registryUrl={{registry.dev.url}}&operation=searchArtifacts")
> .split(body())
> .setHeader("CamelApicurioRegistryOperation",
> constant("getArtifactContent"))
>
> .toD("apicurio-registry://${header.CamelApicurioRegistryGroupId}/${header.CamelApicurioRegistryArtifactId}?registryUrl={{registry.dev.url}}")
> .setHeader("CamelApicurioRegistryOperation",
> constant("createArtifact"))
> .setHeader("CamelApicurioRegistryIfExists",
> constant("CREATE_VERSION"))
>
> .to("apicurio-registry://my-group?registryUrl={{registry.prod.url}}")
> .end()
> .log("Schema migration complete");
> {code}
>
> {*}Use Case 5: AsyncAPI-Driven Event Router{*}{*}{{*}}
> Use AsyncAPI specs stored in the registry to dynamically configure Camel
> routes for event-driven architectures — the registry becomes the source of
> truth for messaging topology.
> {code:java}
> // Read an AsyncAPI spec and route events to the correct channel
> from("timer:configRefresh?period=120000")
> .setHeader("CamelApicurioRegistryOperation",
> constant("getArtifactContent"))
>
> .to("apicurio-registry://event-platform/order-events?registryUrl={{registry.url}}&artifactType=ASYNCAPI")
> .bean("asyncApiRouteBuilder", "updateRoutes");
>
> {code}
> {*}Use Case 6: Schema Compatibility Gate in CI/CD{*}{*}{{*}}
> Integrate into a Tekton/Jenkins pipeline to test schema compatibility before
> merging changes. Fail the build if a new schema version would break consumers.
> {code:java}
> // CI/CD compatibility check
> from("direct:checkCompatibility")
> .setHeader("CamelApicurioRegistryOperation",
> constant("testCompatibility"))
>
> .to("apicurio-registry://my-group/user-event?registryUrl={{registry.url}}")
> .choice()
> .when(header("CamelApicurioRegistryCompatible").isEqualTo(true))
> .log("Schema is compatible — safe to deploy")
> .to("direct:proceed")
> .otherwise()
> .log("BREAKING CHANGE detected!")
> .to("direct:failBuild")
> .end();
>
> {code}
> {*}Use Case 7: Data Format — Schema-Validated Marshalling{*}{*}{{*}}
> Use Apicurio Registry as a Camel DataFormat for serialization/deserialization
> in any route (not just Kafka), pulling schemas from the registry at runtime.
>
> {code:java}
> // Unmarshal JMS messages using Avro schema from the registry
> from("jms:queue:orders")
> .unmarshal().custom("apicurio-avro:my-group/order-schema")
> .process(exchange -> {
> GenericRecord order = exchange.getIn().getBody(GenericRecord.class);
> // process the order...
> })
> .marshal().custom("apicurio-avro:my-group/order-response-schema")
> .to("jms:queue:order-responses");
> {code}
> {*}Supported Artifact Types{*}{*}{{*}}
> Apicurio Registry supports a wide range of artifact types beyond traditional
> schemas, making the component useful across many integration scenarios.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
