This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new fd6bd8013a66 CAMEL-25807: camel-github2 - the commit consumer no
longer loses the commits beyond the newest 100 of a poll (#28199)
fd6bd8013a66 is described below
commit fd6bd8013a66d6c3460eb170a59afc3a572f0b6c
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 22 00:38:44 2026 +0530
CAMEL-25807: camel-github2 - the commit consumer no longer loses the
commits beyond the newest 100 of a poll (#28199)
* CAMEL-25807: camel-github2 - the commit consumer no longer loses the
commits beyond the newest 100 of a poll
A poll of the commit consumer read the 100 newest commits of the branch and
routed those after the last routed commit, which then became the newest one.
When more than 100 commits were pushed between two polls (a large merge, or
the first poll after GitHub was unreachable for a while), the last routed
commit was not among them and the older new commits were never routed.
The poll now reads up to the last routed commit (at most 1000, the number of
commits the consumer remembers). When it is not on the branch any more, only
the 100 newest commits are considered, as before; when none of them is
new, the consumer continues from the newest one, so that the next polls
do not read up to 1000 commits again to look for the last routed one.
* CAMEL-25807: camel-github2 - log when the last routed commit is no longer
on the branch
* CAMEL-25807: camel-github2 - address review: WARN when the last routed
commit is not among the newest 1000
When the listing reaches its capacity without finding the last routed
commit, more than 999 commits may have been pushed since the last poll,
and the ones before the newest 100 are skipped: the consumer now logs a
WARN for it. The comment above the loop says that such a commit can also
still be on the branch, further back.
* CAMEL-25807: camel-github2 - address review: merged-in commits and the
API cost of a catch-up poll
GitHub lists the commits of a branch by committer date, so reading up to
the last routed commit covers a linear history only: the commits that a
merge brings in keep their older dates and can be listed after it. The
comment and the upgrade guide now say so, and the heading says that the
consumer routes the commits beyond the newest 100 rather than all of
them. The upgrade guide also gives the API cost of a poll that catches up
on many commits (one request per routed commit for its author) and says
that the GitHub client waits inside the poll at the rate limit.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/github2/consumer/CommitConsumer.java | 33 +++-
.../GitHub2CommitConsumerManyNewCommitsTest.java | 179 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 ++
3 files changed, 224 insertions(+), 2 deletions(-)
diff --git
a/components/camel-github2/src/main/java/org/apache/camel/component/github2/consumer/CommitConsumer.java
b/components/camel-github2/src/main/java/org/apache/camel/component/github2/consumer/CommitConsumer.java
index a408c3ad7e9f..2db3161f658f 100644
---
a/components/camel-github2/src/main/java/org/apache/camel/component/github2/consumer/CommitConsumer.java
+++
b/components/camel-github2/src/main/java/org/apache/camel/component/github2/consumer/CommitConsumer.java
@@ -37,6 +37,7 @@ public class CommitConsumer extends AbstractGitHub2Consumer {
private static final Logger LOG =
LoggerFactory.getLogger(CommitConsumer.class);
private static final int CAPACITY = 1000;
+ private static final int MAX_COMMITS_WITHOUT_LAST_SHA = 100;
private final String branchName;
private final String startingSha;
@@ -109,12 +110,32 @@ public class CommitConsumer extends
AbstractGitHub2Consumer {
}
List<GHCommit> commits = new ArrayList<>();
+ // read up to the last routed commit, and not only the first 100:
lastSha then moves to the newest one, so
+ // the older new commits would be lost when more than 100 were
pushed between two polls. GitHub lists the
+ // commits by committer date, newest first: in a linear history
the commits pushed since the last routed
+ // one all come before it, but the commits that a merge brings in
keep their older dates and can come after
+ // it, and are not routed. Without a last commit, or when it is
not among the remembered number of newest
+ // commits (it is no longer on the branch, or it is further back
than that), only the 100 newest are
+ // considered
+ int limit = lastSha != null ? CAPACITY :
MAX_COMMITS_WITHOUT_LAST_SHA;
+ boolean lastShaFound = false;
for (GHCommit commit :
getRepository().queryCommits().from(branchName).list().withPageSize(100)) {
commits.add(commit);
- // Stop after 100 commits to avoid fetching too many
- if (commits.size() >= 100) {
+ if (lastSha != null && lastSha.equals(commit.getSHA1())) {
+ lastShaFound = true;
break;
}
+ if (commits.size() >= limit) {
+ break;
+ }
+ }
+ if (lastSha != null && !lastShaFound && commits.size() >=
CAPACITY) {
+ LOG.warn("Last routed commit {} not found among the newest {}
commits of branch {}: routing the newest {}"
+ + " only. If it is still on the branch, the new
commits before those are skipped",
+ lastSha, CAPACITY, branchName,
MAX_COMMITS_WITHOUT_LAST_SHA);
+ }
+ if (!lastShaFound && commits.size() >
MAX_COMMITS_WITHOUT_LAST_SHA) {
+ commits = commits.subList(0, MAX_COMMITS_WITHOUT_LAST_SHA);
}
// clip the list after the last sha
@@ -140,6 +161,14 @@ public class CommitConsumer extends
AbstractGitHub2Consumer {
newCommits.push(commit);
}
}
+ if (newCommits.isEmpty() && lastSha != null && !lastShaFound &&
!commits.isEmpty()) {
+ // the last routed commit is no longer on the branch (a force
push or a reset to an older commit), and
+ // the newest commits were all routed before: continue from
the newest one, so that the next polls do
+ // not read the remembered number of commits again to look for
the last routed one
+ LOG.info("Last routed commit {} is no longer on branch {},
continuing from the newest commit {}",
+ lastSha, branchName, commits.get(0).getSHA1());
+ lastSha = commits.get(0).getSHA1();
+ }
Queue<Object> exchanges = new ArrayDeque<>();
while (!newCommits.isEmpty()) {
diff --git
a/components/camel-github2/src/test/java/org/apache/camel/component/github2/consumer/GitHub2CommitConsumerManyNewCommitsTest.java
b/components/camel-github2/src/test/java/org/apache/camel/component/github2/consumer/GitHub2CommitConsumerManyNewCommitsTest.java
new file mode 100644
index 000000000000..e7c690ca484c
--- /dev/null
+++
b/components/camel-github2/src/test/java/org/apache/camel/component/github2/consumer/GitHub2CommitConsumerManyNewCommitsTest.java
@@ -0,0 +1,179 @@
+/*
+ * 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.github2.consumer;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.github2.GitHub2Constants;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.kohsuke.github.GHCommit;
+import org.kohsuke.github.GHCommitQueryBuilder;
+import org.kohsuke.github.GHRepository;
+import org.kohsuke.github.GitHub;
+import org.kohsuke.github.PagedIterable;
+import org.kohsuke.github.PagedIterator;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * More than 100 commits pushed between two polls (a large merge, or the first
poll after GitHub was unreachable for a
+ * while): every one of them must be routed, not only the newest 100.
+ */
+class GitHub2CommitConsumerManyNewCommitsTest extends CamelTestSupport {
+
+ private static final int NEW_COMMITS = 150;
+
+ // newest first, as the API lists them; c0 is the commit the consumer
starts after
+ private volatile List<GHCommit> commits = createCommits(true);
+
+ @BindToRegistry(GitHub2Constants.GITHUB_CLIENT)
+ private final GitHub github = createGitHub();
+
+ private final List<String> routed = new CopyOnWriteArrayList<>();
+ // the number of commits read by each listing of the branch, one entry per
listing
+ private final List<AtomicInteger> listings = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("github2:commit/main?startingSha=c0&repoOwner=acme&repoName=demo&oauthToken=not-used"
+ + "&initialDelay=0&delay=10")
+ .routeId("commit").autoStartup(false)
+ .process(e ->
routed.add(e.getMessage().getHeader(GitHub2Constants.GITHUB_COMMIT_SHA,
String.class)));
+ }
+ };
+ }
+
+ @Test
+ void everyNewCommitIsRouted() throws Exception {
+ context.getRouteController().startRoute("commit");
+ List<String> expected = new ArrayList<>();
+ for (int i = 1; i <= NEW_COMMITS; i++) {
+ expected.add("c" + i);
+ }
+ // the polls after the first one find no new commit; give them time to
run
+ await().pollDelay(500, TimeUnit.MILLISECONDS).atMost(10,
TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(expected, routed));
+ }
+
+ @Test
+ void withoutTheLastCommitOnTheBranchOnlyTheNewest100AreRouted() throws
Exception {
+ // c0 is no longer on the branch (a force push): as before, only the
100 newest commits are considered
+ commits = createCommits(false);
+ context.getRouteController().startRoute("commit");
+
+ List<String> expected = new ArrayList<>();
+ for (int i = NEW_COMMITS - 99; i <= NEW_COMMITS; i++) {
+ expected.add("c" + i);
+ }
+ await().pollDelay(500, TimeUnit.MILLISECONDS).atMost(10,
TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(expected, routed));
+ }
+
+ @Test
+ void aBranchResetToRoutedCommitsIsNotListedAgainAtEveryPoll() throws
Exception {
+ context.getRouteController().startRoute("commit");
+ await().atMost(10, TimeUnit.SECONDS).until(() -> routed.size() ==
NEW_COMMITS);
+
+ // the branch is reset to c100: the last routed commit c150 is no
longer on it, and the commits left on it were
+ // all routed before
+ commits = new ArrayList<>(commits.subList(NEW_COMMITS - 100,
commits.size()));
+ int listingsBefore = listings.size();
+
+ // a poll that finds nothing new continues from the newest commit, so
the next polls read one commit only, and
+ // not the whole branch to look for c150 again
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertTrue(listings.size() >= listingsBefore + 3, "polls after the
reset");
+ assertEquals(1, listings.get(listings.size() - 2).get(), "commits
read by a poll after the reset");
+ });
+ assertEquals(NEW_COMMITS, routed.size(), "commits routed");
+ }
+
+ private static List<GHCommit> createCommits(boolean withStartingCommit) {
+ try {
+ List<GHCommit> answer = new ArrayList<>();
+ for (int i = NEW_COMMITS; i >= (withStartingCommit ? 0 : 1); i--) {
+ answer.add(commit("c" + i));
+ }
+ return answer;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private GitHub createGitHub() {
+ try {
+ GHRepository repository = mock(GHRepository.class);
+ GHCommitQueryBuilder commitQuery =
mock(GHCommitQueryBuilder.class);
+ when(commitQuery.from(anyString())).thenReturn(commitQuery);
+ when(commitQuery.list()).thenAnswer(invocation -> paged(commits));
+ when(repository.queryCommits()).thenReturn(commitQuery);
+
+ GitHub client = mock(GitHub.class);
+ when(client.getRepository("acme/demo")).thenReturn(repository);
+ return client;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private <T> PagedIterable<T> paged(List<T> items) {
+ return new PagedIterable<>() {
+ @Override
+ public PagedIterator<T> _iterator(int pageSize) {
+ Iterator<T> delegate = items.iterator();
+ AtomicInteger read = new AtomicInteger();
+ listings.add(read);
+ PagedIterator<T> iterator = mock(PagedIterator.class);
+ when(iterator.hasNext()).thenAnswer(invocation ->
delegate.hasNext());
+ when(iterator.next()).thenAnswer(invocation -> {
+ read.incrementAndGet();
+ return delegate.next();
+ });
+ return iterator;
+ }
+ };
+ }
+
+ private static GHCommit commit(String sha) throws IOException {
+ GHCommit commit = mock(GHCommit.class, RETURNS_DEEP_STUBS);
+ when(commit.getSHA1()).thenReturn(sha);
+ when(commit.getAuthor()).thenReturn(null);
+ when(commit.getCommitter()).thenReturn(null);
+
when(commit.getHtmlUrl()).thenReturn(URI.create("https://github.invalid/acme/demo/commit/";
+ sha).toURL());
+ when(commit.getCommitShortInfo().getMessage()).thenReturn("message of
" + sha);
+ return commit;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 3d848a8480c8..273f5d2b8a8e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1133,6 +1133,20 @@ file is written from the column names, not from the
first map). Every map of the
`ignoreFirstRecord` only applies to unmarshal. A route that set
`ignoreFirstRecord=false` to work around this is not
affected; a route that added a dummy first map now gets it in the output too.
+=== camel-github2 - the commit consumer routes the commits beyond the newest
100
+
+The `commit` consumer read only the 100 newest commits of the branch at each
poll, so when more than 100 commits were
+pushed between two polls, the older ones were never routed. A poll now reads
the commits up to the last routed one
+(at most 1000, the number of commits the consumer remembers), and can route
more than 100 commits. When the last
+routed commit is no longer on the branch, only the 100 newest commits are
considered, as before. GitHub lists the
+commits by committer date, newest first, so this covers a linear history: the
commits that a merge brings in keep
+their older dates and can be listed after the last routed commit, and they are
not routed, as before.
+
+Each routed commit costs one GitHub API request for its author, plus one for
each author or committer not looked up
+before, so a poll that catches up on many commits can send about 1000 requests
instead of about 100. When the rate
+limit is reached, the GitHub client waits inside the poll until the limit
resets (its default rate limit handler),
+and stopping the route waits for that poll.
+
=== camel-milo - the OPC UA client validates the server certificate
When the endpoint it selects uses a security policy other than `None`, the
`milo-client` and `milo-browse` components
