oscerd commented on code in PR #24844: URL: https://github.com/apache/camel/pull/24844#discussion_r4193911194
########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryProducer.java: ########## @@ -0,0 +1,275 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.apicurioregistry; + +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; + +import io.apicurio.registry.rest.client.RegistryClient; +import io.apicurio.registry.rest.client.models.ArtifactMetaData; +import io.apicurio.registry.rest.client.models.CreateArtifact; +import io.apicurio.registry.rest.client.models.CreateArtifactResponse; +import io.apicurio.registry.rest.client.models.CreateGroup; +import io.apicurio.registry.rest.client.models.CreateVersion; +import io.apicurio.registry.rest.client.models.GroupMetaData; +import io.apicurio.registry.rest.client.models.IfArtifactExists; +import io.apicurio.registry.rest.client.models.RuleViolationProblemDetails; +import io.apicurio.registry.rest.client.models.SearchedVersion; +import io.apicurio.registry.rest.client.models.SortOrder; +import io.apicurio.registry.rest.client.models.VersionContent; +import io.apicurio.registry.rest.client.models.VersionMetaData; +import io.apicurio.registry.rest.client.models.VersionSearchResults; +import io.apicurio.registry.rest.client.models.VersionSortBy; +import org.apache.camel.Message; +import org.apache.camel.spi.InvokeOnHeader; +import org.apache.camel.support.HeaderSelectorProducer; + +public class ApicurioRegistryProducer extends HeaderSelectorProducer { + + private static final int PAGE_SIZE = 100; + + private final ApicurioRegistryEndpoint endpoint; + private final ApicurioRegistryConfiguration configuration; + + public ApicurioRegistryProducer(ApicurioRegistryEndpoint endpoint, + ApicurioRegistryConfiguration configuration) { + super(endpoint, ApicurioRegistryConstants.HEADER_OPERATION, configuration::getOperation); + this.endpoint = endpoint; + this.configuration = configuration; + } + + private RegistryClient getClient() { + return endpoint.getRegistryClient(); + } + + private String resolveGroupId(Message message) { + String gid = message.getHeader(ApicurioRegistryConstants.HEADER_GROUP_ID, String.class); + return gid != null ? gid : endpoint.getGroupId(); + } + + private String resolveArtifactId(Message message) { + String aid = message.getHeader(ApicurioRegistryConstants.HEADER_ARTIFACT_ID, String.class); + return aid != null ? aid : endpoint.getArtifactId(); + } + + @InvokeOnHeader(ApicurioRegistryConstants.OPERATION_CREATE_ARTIFACT) + public void createArtifact(Message message) { + String groupId = resolveGroupId(message); + String artifactId = resolveArtifactId(message); + String artifactType = message.getHeader( + ApicurioRegistryConstants.HEADER_ARTIFACT_TYPE, configuration.getArtifactType(), String.class); + String name = message.getHeader(ApicurioRegistryConstants.HEADER_ARTIFACT_NAME, String.class); + String description = message.getHeader(ApicurioRegistryConstants.HEADER_ARTIFACT_DESCRIPTION, String.class); + String content = message.getBody(String.class); + String contentType = message.getHeader( + ApicurioRegistryConstants.HEADER_CONTENT_TYPE, "application/json", String.class); + String ifExistsVal = message.getHeader( + ApicurioRegistryConstants.HEADER_IF_EXISTS, configuration.getIfExists(), String.class); + + CreateArtifact createArtifact = new CreateArtifact(); + createArtifact.setArtifactId(artifactId); + createArtifact.setArtifactType(artifactType); + createArtifact.setName(name); + createArtifact.setDescription(description); + + if (content != null) { + CreateVersion firstVersion = new CreateVersion(); + VersionContent vc = new VersionContent(); + vc.setContent(content); + vc.setContentType(contentType); + firstVersion.setContent(vc); + createArtifact.setFirstVersion(firstVersion); + } + + CreateArtifactResponse result = getClient().groups().byGroupId(groupId).artifacts() + .post(createArtifact, config -> { + if (ifExistsVal != null) { + config.queryParameters.ifExists = IfArtifactExists.forValue(ifExistsVal); + } + }); + message.setBody(result); + } + + @InvokeOnHeader(ApicurioRegistryConstants.OPERATION_UPDATE_ARTIFACT) + public void updateArtifact(Message message) { + String groupId = resolveGroupId(message); + String artifactId = resolveArtifactId(message); + String content = message.getBody(String.class); + String version = message.getHeader(ApicurioRegistryConstants.HEADER_VERSION, String.class); + String contentType = message.getHeader( + ApicurioRegistryConstants.HEADER_CONTENT_TYPE, "application/json", String.class); + + CreateVersion createVersion = new CreateVersion(); + createVersion.setVersion(version); + VersionContent vc = new VersionContent(); + vc.setContent(content); + vc.setContentType(contentType); + createVersion.setContent(vc); + + VersionMetaData result = getClient().groups().byGroupId(groupId).artifacts() + .byArtifactId(artifactId).versions().post(createVersion); + message.setBody(result); + } + + @InvokeOnHeader(ApicurioRegistryConstants.OPERATION_DELETE_ARTIFACT) + public void deleteArtifact(Message message) { + String groupId = resolveGroupId(message); + String artifactId = resolveArtifactId(message); + getClient().groups().byGroupId(groupId).artifacts().byArtifactId(artifactId).delete(); + } + + @InvokeOnHeader(ApicurioRegistryConstants.OPERATION_GET_ARTIFACT_CONTENT) + public void getArtifactContent(Message message) throws Exception { + String groupId = resolveGroupId(message); + String artifactId = resolveArtifactId(message); + String version = message.getHeader( + ApicurioRegistryConstants.HEADER_VERSION, "branch=latest", String.class); + + try (InputStream content = getClient().groups().byGroupId(groupId).artifacts() + .byArtifactId(artifactId).versions().byVersionExpression(version).content().get()) { + message.setBody(content.readAllBytes()); + } + } + + @InvokeOnHeader(ApicurioRegistryConstants.OPERATION_GET_ARTIFACT_METADATA) + public void getArtifactMetadata(Message message) { + String groupId = resolveGroupId(message); + String artifactId = resolveArtifactId(message); + + ArtifactMetaData metadata = getClient().groups().byGroupId(groupId).artifacts() + .byArtifactId(artifactId).get(); + message.setBody(metadata); + } + + @InvokeOnHeader(ApicurioRegistryConstants.OPERATION_SEARCH_ARTIFACTS) + public void searchArtifacts(Message message) { + var results = getClient().search().artifacts().get(config -> { Review Comment: **searchArtifacts returns only the registry's first page (20) and ignores the artifact ID** Only `name`, `groupId` and `description` are set, so the registry falls back to `offset=0`/`limit=20`: a group with more than 20 artifacts comes back silently truncated, and no header lets you ask for the next page (the same truncation that `listVersions` was just fixed for). The artifact ID (header or URI path) is also never applied as a filter; `ApicurioRegistryProducerIT#testSearchArtifacts` puts `CamelApicurioRegistryArtifactId` in its search headers, but the assertion only passes because the random group holds a single artifact. Exposing `offset`/`limit` (and optionally `order`/`orderby`) as headers, mapping the other filters the SDK supports (`artifactId`, `artifactType`, `labels`), and documenting the default page size would cover it. ########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryConsumer.java: ########## @@ -0,0 +1,147 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.apicurioregistry; + +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; + +import io.apicurio.registry.rest.client.RegistryClient; +import io.apicurio.registry.rest.client.models.SearchedVersion; +import io.apicurio.registry.rest.client.models.SortOrder; +import io.apicurio.registry.rest.client.models.VersionSearchResults; +import io.apicurio.registry.rest.client.models.VersionSortBy; +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.Processor; +import org.apache.camel.support.ScheduledPollConsumer; + +public class ApicurioRegistryConsumer extends ScheduledPollConsumer { + + static final int PAGE_SIZE = 100; + + private final ApicurioRegistryEndpoint endpoint; + private final ApicurioRegistryConfiguration configuration; + private volatile Long lastSeenGlobalId; + + public ApicurioRegistryConsumer(ApicurioRegistryEndpoint endpoint, Processor processor, + ApicurioRegistryConfiguration configuration) { + super(endpoint, processor); + this.endpoint = endpoint; + this.configuration = configuration; + } + + @Override + protected int poll() throws Exception { + String groupId = endpoint.getGroupId(); + String artifactId = endpoint.getArtifactId(); + + if (groupId == null || artifactId == null) { + throw new IllegalArgumentException( + "Both groupId and artifactId are required for the consumer"); + } + + RegistryClient client = endpoint.getRegistryClient(); + List<SearchedVersion> versions = fetchNewVersions(client, groupId, artifactId); + + int count = 0; + for (SearchedVersion version : versions) { + Long globalId = version.getGlobalId(); + if (lastSeenGlobalId == null || globalId > lastSeenGlobalId) { + Exchange exchange = createExchange(false); + try { + Message message = exchange.getIn(); + + message.setHeader(ApicurioRegistryConstants.HEADER_GROUP_ID, groupId); + message.setHeader(ApicurioRegistryConstants.HEADER_ARTIFACT_ID, artifactId); + message.setHeader(ApicurioRegistryConstants.HEADER_VERSION, version.getVersion()); + message.setHeader(ApicurioRegistryConstants.HEADER_GLOBAL_ID, globalId); + message.setHeader(ApicurioRegistryConstants.HEADER_CONTENT_ID, version.getContentId()); + message.setHeader(ApicurioRegistryConstants.HEADER_ARTIFACT_TYPE, version.getArtifactType()); + if (version.getState() != null) { + message.setHeader(ApicurioRegistryConstants.HEADER_VERSION_STATE, + version.getState().getValue()); + } + + if (configuration.isFetchContent()) { + try (InputStream content = client.groups().byGroupId(groupId).artifacts() Review Comment: **fetchContent=true stalls forever on a DISABLED version** **A DISABLED version stalls the consumer for good when `fetchContent=true`.** In Registry 3.3.0 the version list returns versions in every state (`GroupsResourceImpl#listArtifactVersions` filters only by group and artifact), but fetching the content of a `DISABLED` version throws `VersionNotFoundException`, a 404 (`GroupsResourceImpl#getArtifactVersionContent`). Here that `ProblemDetails` escapes `poll()` before `lastSeenGlobalId` moves, so every later poll lists the same version first and fails on it again. Example: versions 1 (ENABLED), 2 (DISABLED), 3 (ENABLED): the consumer delivers 1, then logs `Failed polling endpoint ... Will try again at next poll` every `delay`, and never delivers 3 or anything newer. This is reachable on first start through any historically disabled version, or whenever someone retracts a bad version, and because it happens before the route runs, `onException(...).handled(true)` cannot skip it. When `fetchContent=true`, skipping `DISABLED` versions and advancing the watermark past them (or delivering them metadata-only) would fix it, e.g. inside the loop before `createExchange`: ```java if (configuration.isFetchContent() && version.getState() == VersionState.DISABLED) { // the registry returns 404 for the content of a DISABLED version: skip it instead of retrying forever lastSeenGlobalId = globalId; continue; } ``` plus a unit test where `FakeVersionPages` returns a DISABLED version whose `content().get()` throws a 404 `ProblemDetails`. Related, worth a docs note: DRAFT versions are delivered as soon as they are listed, and with `apicurio.rest.mutability.artifact-version-content.enabled` their content can still be edited under the same globalId without being re-delivered. ########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryConsumer.java: ########## @@ -0,0 +1,147 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.apicurioregistry; + +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; + +import io.apicurio.registry.rest.client.RegistryClient; +import io.apicurio.registry.rest.client.models.SearchedVersion; +import io.apicurio.registry.rest.client.models.SortOrder; +import io.apicurio.registry.rest.client.models.VersionSearchResults; +import io.apicurio.registry.rest.client.models.VersionSortBy; +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.Processor; +import org.apache.camel.support.ScheduledPollConsumer; + +public class ApicurioRegistryConsumer extends ScheduledPollConsumer { + + static final int PAGE_SIZE = 100; + + private final ApicurioRegistryEndpoint endpoint; + private final ApicurioRegistryConfiguration configuration; + private volatile Long lastSeenGlobalId; + + public ApicurioRegistryConsumer(ApicurioRegistryEndpoint endpoint, Processor processor, + ApicurioRegistryConfiguration configuration) { + super(endpoint, processor); + this.endpoint = endpoint; + this.configuration = configuration; + } + + @Override + protected int poll() throws Exception { + String groupId = endpoint.getGroupId(); + String artifactId = endpoint.getArtifactId(); + + if (groupId == null || artifactId == null) { Review Comment: **Fail fast on missing group/artifact IDs instead of on every poll** Because the group/artifact check runs inside every `poll()`, a consumer URI without an artifact ID still starts the route and then logs `Failed polling endpoint ... Will try again at next poll` every `delay`, forever; checking in `ApicurioRegistryEndpoint#createConsumer` would fail at startup. On the producer side, an operation that needs a missing ID gets a message-less `NullPointerException` from the SDK's `Objects.requireNonNull`; a per-operation check with a clear `IllegalArgumentException` would help. ########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryConfiguration.java: ########## @@ -0,0 +1,188 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.apicurioregistry; + +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriParams; + +@UriParams +public class ApicurioRegistryConfiguration implements Cloneable { + + @UriParam(label = "common", description = "The Apicurio Registry base URL") + @Metadata(required = true) + private String registryUrl; + + @UriParam(label = "producer", + enums = "createArtifact,updateArtifact,deleteArtifact,getArtifactContent,getArtifactMetadata,searchArtifacts,listVersions,createGroup,testCompatibility,validate", + description = "The operation to perform") + private String operation; + + @UriParam(label = "producer", description = "The default artifact type", defaultValue = "JSON") + private String artifactType = "JSON"; + + @UriParam(label = "security", enums = "none,basic,oidc", defaultValue = "none", + description = "The authentication type to use") + private String authType = "none"; + + @UriParam(label = "security", description = "Username for basic authentication") + private String username; + + @UriParam(label = "security", secret = true, description = "Password for basic authentication") Review Comment: **Use security = "secret" instead of the deprecated secret = true** `UriParam#secret()` is deprecated since 4.19.0 in favour of `security = "secret"` (the first revision of this PR used it, as do `camel-opa`/`camel-openfga`); the generated metadata is identical, so no regen is needed. `clientSecret` below needs the same change. ```suggestion @UriParam(label = "security", security = "secret", description = "Password for basic authentication") ``` ########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryConsumer.java: ########## @@ -0,0 +1,147 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.apicurioregistry; + +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; + +import io.apicurio.registry.rest.client.RegistryClient; +import io.apicurio.registry.rest.client.models.SearchedVersion; +import io.apicurio.registry.rest.client.models.SortOrder; +import io.apicurio.registry.rest.client.models.VersionSearchResults; +import io.apicurio.registry.rest.client.models.VersionSortBy; +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.Processor; +import org.apache.camel.support.ScheduledPollConsumer; + +public class ApicurioRegistryConsumer extends ScheduledPollConsumer { + + static final int PAGE_SIZE = 100; + + private final ApicurioRegistryEndpoint endpoint; + private final ApicurioRegistryConfiguration configuration; + private volatile Long lastSeenGlobalId; + + public ApicurioRegistryConsumer(ApicurioRegistryEndpoint endpoint, Processor processor, + ApicurioRegistryConfiguration configuration) { + super(endpoint, processor); + this.endpoint = endpoint; + this.configuration = configuration; + } + + @Override + protected int poll() throws Exception { + String groupId = endpoint.getGroupId(); + String artifactId = endpoint.getArtifactId(); + + if (groupId == null || artifactId == null) { + throw new IllegalArgumentException( + "Both groupId and artifactId are required for the consumer"); + } + + RegistryClient client = endpoint.getRegistryClient(); + List<SearchedVersion> versions = fetchNewVersions(client, groupId, artifactId); + + int count = 0; + for (SearchedVersion version : versions) { + Long globalId = version.getGlobalId(); + if (lastSeenGlobalId == null || globalId > lastSeenGlobalId) { + Exchange exchange = createExchange(false); + try { + Message message = exchange.getIn(); + + message.setHeader(ApicurioRegistryConstants.HEADER_GROUP_ID, groupId); + message.setHeader(ApicurioRegistryConstants.HEADER_ARTIFACT_ID, artifactId); + message.setHeader(ApicurioRegistryConstants.HEADER_VERSION, version.getVersion()); + message.setHeader(ApicurioRegistryConstants.HEADER_GLOBAL_ID, globalId); + message.setHeader(ApicurioRegistryConstants.HEADER_CONTENT_ID, version.getContentId()); + message.setHeader(ApicurioRegistryConstants.HEADER_ARTIFACT_TYPE, version.getArtifactType()); + if (version.getState() != null) { + message.setHeader(ApicurioRegistryConstants.HEADER_VERSION_STATE, + version.getState().getValue()); + } + + if (configuration.isFetchContent()) { + try (InputStream content = client.groups().byGroupId(groupId).artifacts() + .byArtifactId(artifactId).versions() + .byVersionExpression(version.getVersion()).content().get()) { + message.setBody(content.readAllBytes()); + } + } else { + message.setBody(version); + } + + getProcessor().process(exchange); + if (exchange.getException() != null) { + // do not advance the watermark so the version is retried on the next poll + getExceptionHandler().handleException( + "Error processing artifact version with globalId " + globalId, exchange, + exchange.getException()); + break; + } + lastSeenGlobalId = globalId; + count++; + } finally { + releaseExchange(exchange, false); + } + } + } + return count; + } + + /** + * Fetches the versions newer than the watermark, in ascending globalId order. Pages are requested newest first so + * that polling stops as soon as an already seen version is reached. + */ + private List<SearchedVersion> fetchNewVersions(RegistryClient client, String groupId, String artifactId) { + // keyed by globalId: a version created between page requests shifts the pages and may repeat one entry + Map<Long, SearchedVersion> answer = new TreeMap<>(); + int offset = 0; + while (true) { + final int pageOffset = offset; + VersionSearchResults page = client.groups().byGroupId(groupId).artifacts().byArtifactId(artifactId) + .versions().get(config -> { + config.queryParameters.orderby = VersionSortBy.GlobalId; + config.queryParameters.order = SortOrder.Desc; + config.queryParameters.offset = pageOffset; + config.queryParameters.limit = PAGE_SIZE; Review Comment: **Idle polls download up to 100 versions every 500 ms** Every poll starts with a `limit=100` page, newest first, although an idle poll only needs to see that the newest globalId is not above the watermark; with the inherited 500 ms `delay`, an artifact with 100+ versions costs a 100-entry page twice a second per consumer against what is usually a shared registry. A `limit=1` probe first and/or a larger default `delay` (as `MailConsumer`/`FeedPollingConsumer` or `SlackConsumer` do) would be kinder to the registry. ########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryProducer.java: ########## @@ -0,0 +1,275 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.apicurioregistry; + +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; + +import io.apicurio.registry.rest.client.RegistryClient; +import io.apicurio.registry.rest.client.models.ArtifactMetaData; +import io.apicurio.registry.rest.client.models.CreateArtifact; +import io.apicurio.registry.rest.client.models.CreateArtifactResponse; +import io.apicurio.registry.rest.client.models.CreateGroup; +import io.apicurio.registry.rest.client.models.CreateVersion; +import io.apicurio.registry.rest.client.models.GroupMetaData; +import io.apicurio.registry.rest.client.models.IfArtifactExists; +import io.apicurio.registry.rest.client.models.RuleViolationProblemDetails; +import io.apicurio.registry.rest.client.models.SearchedVersion; +import io.apicurio.registry.rest.client.models.SortOrder; +import io.apicurio.registry.rest.client.models.VersionContent; +import io.apicurio.registry.rest.client.models.VersionMetaData; +import io.apicurio.registry.rest.client.models.VersionSearchResults; +import io.apicurio.registry.rest.client.models.VersionSortBy; +import org.apache.camel.Message; +import org.apache.camel.spi.InvokeOnHeader; +import org.apache.camel.support.HeaderSelectorProducer; + +public class ApicurioRegistryProducer extends HeaderSelectorProducer { + + private static final int PAGE_SIZE = 100; + + private final ApicurioRegistryEndpoint endpoint; + private final ApicurioRegistryConfiguration configuration; + + public ApicurioRegistryProducer(ApicurioRegistryEndpoint endpoint, + ApicurioRegistryConfiguration configuration) { + super(endpoint, ApicurioRegistryConstants.HEADER_OPERATION, configuration::getOperation); + this.endpoint = endpoint; + this.configuration = configuration; + } + + private RegistryClient getClient() { + return endpoint.getRegistryClient(); + } + + private String resolveGroupId(Message message) { + String gid = message.getHeader(ApicurioRegistryConstants.HEADER_GROUP_ID, String.class); + return gid != null ? gid : endpoint.getGroupId(); Review Comment: **Consumer headers override a downstream producer's URI path** The consumer sets the group/artifact/version headers on every exchange, and the producer prefers those headers over its endpoint path (and `updateArtifact` sends the version header). So in a promotion route like `from("apicurio-registry:dev/orders?fetchContent=true").to("apicurio-registry:prod/orders?operation=updateArtifact")` the producer targets `dev/orders` with an already existing version and gets a 409. At least a docs note (`removeHeaders("CamelApicurioRegistry*")` in between) would help; camel-kafka avoids it by only honouring an explicit override header that it removes after use. ########## components/camel-apicurio-registry/src/main/docs/apicurio-registry-component.adoc: ########## @@ -0,0 +1,241 @@ += Apicurio Registry Component +:doctitle: Apicurio Registry +:shortname: apicurio-registry +:artifactid: camel-apicurio-registry +:description: Manage artifacts, versions, and groups in Apicurio Registry v3. +:since: 4.23 +:supportlevel: Preview +:tabs-sync-option: +:component-header: Both producer and consumer are supported + +*Since Camel {since}* + +*{component-header}* + +The Apicurio Registry component provides integration with https://www.apicur.io/registry/[Apicurio Registry] v3 +using the https://github.com/Apicurio/apicurio-registry[Apicurio Registry Java SDK]. + +It supports CRUD operations on artifacts, groups, and versions, as well as schema validation +and polling for new artifact versions. + +Maven users will need to add the following dependency to their `pom.xml` +for this component: + +[source,xml] +---- +<dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-apicurio-registry</artifactId> + <version>x.x.x</version> + <!-- use the same version as your Camel core version --> +</dependency> +---- + +== URI format + +---- +apicurio-registry:groupId/artifactId[?options] +---- + +Where `groupId` and `artifactId` identify the registry artifact. Both are optional for +operations that do not require them (e.g., `searchArtifacts`, `createGroup`). + +The first `/` separates the group ID from the artifact ID. Group IDs containing `/` +cannot be represented in the URI path. Producers can instead supply the group ID using +the `CamelApicurioRegistryGroupId` header, which overrides the URI value. + +// component options: START +include::partial$component-configure-options.adoc[] +include::partial$component-endpoint-options.adoc[] +include::partial$component-endpoint-headers.adoc[] +// component options: END + +== Producer Operations + +The producer supports the following operations, set via the `operation` endpoint option or +the `CamelApicurioRegistryOperation` header: + +[cols="1,3"] +|=== +| Operation | Description + +| `createArtifact` | Create a new artifact. Body contains the artifact content. +| `updateArtifact` | Create a new version of an existing artifact. Body contains the new content. +| `deleteArtifact` | Delete an artifact. Review Comment: **Document that the registry must allow artifact deletion** Registry 3.x ships with artifact deletion disabled (`apicurio.rest.deletion.artifact.enabled=false`), so `deleteArtifact` fails with a 405 on a default registry; the ITs pass only because the test infra sets `APICURIO_REST_DELETION_ARTIFACT_ENABLED=true`. A short note here would save users some confusion (and regenerate the catalog doc mirror after editing). ########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryEndpoint.java: ########## @@ -0,0 +1,139 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.apicurioregistry; + +import io.apicurio.registry.client.RegistryClientFactory; +import io.apicurio.registry.client.common.RegistryClientOptions; +import io.apicurio.registry.rest.client.RegistryClient; +import org.apache.camel.Category; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.spi.EndpointServiceLocation; +import org.apache.camel.spi.UriEndpoint; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriPath; +import org.apache.camel.support.ScheduledPollEndpoint; + +/** + * Manage artifacts, versions, and groups in Apicurio Registry v3. + */ +@UriEndpoint(firstVersion = "4.23.0", scheme = "apicurio-registry", title = "Apicurio Registry", + syntax = "apicurio-registry:groupId/artifactId", + category = { Category.CLOUD, Category.API }, headersClass = ApicurioRegistryConstants.class) +public class ApicurioRegistryEndpoint extends ScheduledPollEndpoint implements EndpointServiceLocation { + + @UriPath(description = "The artifact group ID") + private String groupId; + + @UriPath(description = "The artifact ID") + private String artifactId; + + @UriParam + private ApicurioRegistryConfiguration configuration; + + @UriParam(label = "advanced", description = "To use a pre-configured RegistryClient instance") + private RegistryClient registryClient; + + private boolean managedRegistryClient; + + ApicurioRegistryEndpoint(String uri, ApicurioRegistryComponent component, + ApicurioRegistryConfiguration configuration, + String groupId, String artifactId) { + super(uri, component); + this.configuration = configuration; + this.groupId = groupId; + this.artifactId = artifactId; + } + + @Override + public Producer createProducer() throws Exception { + return new ApicurioRegistryProducer(this, configuration); + } + + @Override + public Consumer createConsumer(Processor processor) throws Exception { + ApicurioRegistryConsumer consumer = new ApicurioRegistryConsumer(this, processor, configuration); + configureConsumer(consumer); + return consumer; + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + if (registryClient == null) { + registryClient = createRegistryClient(); + managedRegistryClient = true; + } + } + + @Override + protected void doStop() throws Exception { + super.doStop(); + if (managedRegistryClient) { + registryClient = null; + managedRegistryClient = false; + } + } + + private RegistryClient createRegistryClient() { + RegistryClientOptions options = RegistryClientOptions.create(configuration.getRegistryUrl(), Review Comment: **No TLS (trust store / mTLS) or proxy configuration** Only the URL, Vert.x and auth are passed to `RegistryClientOptions`; SDK 3.3.0 also supports trust stores, client keystores for mTLS and proxies. Today an HTTPS registry behind a private CA or requiring client certificates can only be reached through the JVM-wide trust store or a hand-built `registryClient`, and the docs mention neither. The defaults are safe (`trustAll=false`, `verifyHost=true`), so this is a gap, not a vulnerability; if `trustAll`/`verifyHost` are ever exposed they need `security = "insecure:ssl"`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
