gnodet-bot commented on code in PR #24844: URL: https://github.com/apache/camel/pull/24844#discussion_r4025083176
########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryConsumer.java: ########## @@ -0,0 +1,105 @@ +/* + * 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.Comparator; +import java.util.List; + +import io.apicurio.registry.rest.client.RegistryClient; +import io.apicurio.registry.rest.client.models.SearchedVersion; +import io.apicurio.registry.rest.client.models.VersionSearchResults; +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 { + + 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(); + VersionSearchResults results = client.groups().byGroupId(groupId) + .artifacts().byArtifactId(artifactId).versions().get(); Review Comment: **[Re-raised from `5a4e13f66c4`] Unbounded `versions().get()` — pagination gap, not just OOM risk.** The Apicurio Registry SDK's `VersionsRequestBuilder.get()` accepts optional `offset` and `limit` query parameters. Without them, the registry returns its default page (typically capped at 20 results). This means: 1. **On first start**: an artifact with >20 existing versions will only emit the first page. The consumer advances `lastSeenGlobalId` past those, so versions 21+ are never processed on this start. 2. **On steady-state polling**: each new poll also fetches only the first page. Once the watermark is high enough that all versions on the first page are below it, the loop does nothing — but new versions appended beyond page 1 are silently dropped until they fall within the returned page. The fix is to paginate or, more pragmatically, filter server-side by `offset`/`limit` once `lastSeenGlobalId` is known: ```suggestion VersionSearchResults results = client.groups().byGroupId(groupId) .artifacts().byArtifactId(artifactId).versions().get( req -> { req.queryParameters.orderby = "globalId"; req.queryParameters.order = "asc"; req.queryParameters.offset = 0; req.queryParameters.limit = 500; }); ``` If the SDK does not expose `orderby`/`order` on `VersionsRequestBuilderGetQueryParameters`, omit them and keep the existing sort. The key fix is to raise `limit` beyond the default so that in practice all versions are fetched. For correctness at scale, loop with increasing `offset` while `results.getCount() > offset + limit`. ########## components/camel-apicurio-registry/src/main/java/org/apache/camel/component/apicurioregistry/ApicurioRegistryEndpoint.java: ########## @@ -0,0 +1,154 @@ +/* + * 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.util.concurrent.TimeUnit; + +import io.apicurio.registry.client.RegistryClientFactory; +import io.apicurio.registry.client.common.RegistryClientOptions; +import io.apicurio.registry.rest.client.RegistryClient; +import io.vertx.core.Vertx; +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 Vertx vertx; + + 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) { + vertx = Vertx.vertx(); + try { + registryClient = createRegistryClient(); + } catch (Exception e) { + closeVertx(); + throw e; + } + } + } + + @Override + protected void doStop() throws Exception { + super.doStop(); + if (vertx != null) { + registryClient = null; + closeVertx(); + } + } + + private void closeVertx() throws Exception { + try { + vertx.close().toCompletionStage().toCompletableFuture().get(30, TimeUnit.SECONDS); + } finally { + vertx = null; + } + } + + private RegistryClient createRegistryClient() { + RegistryClientOptions options = RegistryClientOptions.create(configuration.getRegistryUrl(), vertx); + String authType = configuration.getAuthType(); + if ("basic".equalsIgnoreCase(authType)) { + options.basicAuth(configuration.getUsername(), configuration.getPassword()); + } else if ("oidc".equalsIgnoreCase(authType)) { + options.oauth2(configuration.getTokenEndpoint(), configuration.getClientId(), + configuration.getClientSecret(), configuration.getScope()); + } + return RegistryClientFactory.create(options); + } + + public RegistryClient getRegistryClient() { + return registryClient; + } + + public void setRegistryClient(RegistryClient registryClient) { + this.registryClient = registryClient; + } + + public String getGroupId() { + return groupId; + } + + public String getArtifactId() { + return artifactId; + } + + public ApicurioRegistryConfiguration getConfiguration() { + return configuration; + } + + public void setConfiguration(ApicurioRegistryConfiguration configuration) { + this.configuration = configuration; + } + + @Override + public String getServiceUrl() { + return configuration.getRegistryUrl(); + } + + @Override + public String getServiceProtocol() { + return "http"; Review Comment: **[Re-raised from `5a4e13f66c4`] `getServiceProtocol()` returns `"http"` — misleading service location metadata.** `EndpointServiceLocation` is used by Camel's observability layer to categorise and display service connections. Returning `"http"` makes the registry appear as a plain HTTP endpoint rather than a schema registry, losing the component identity. Other REST-based components use a logical name — e.g. `"elasticsearch"`, `"influxdb"`, `"solr"`. Return a meaningful protocol identifier: ```suggestion public String getServiceProtocol() { return "apicurio-registry"; } ``` The test in `ApicurioRegistryEndpointTest` that asserts `isEqualTo("http")` will need updating to match. -- 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]
