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 3651230156ec CAMEL-26651, CAMEL-26652: camel-git - consumer honours
scheduler options and redelivers failed items (#29270)
3651230156ec is described below
commit 3651230156ecaa4659333ef3b857dc89762a74fe
Author: Federico Mariani <[email protected]>
AuthorDate: Fri Nov 6 08:21:28 2026 +0100
CAMEL-26651, CAMEL-26652: camel-git - consumer honours scheduler options
and redelivers failed items (#29270)
The git consumer honours the scheduler options (greedy, backoff,
sendEmptyMessageWhenIdle) and redelivers items whose processing failed, and
opens the repository before the first poll can run.
AI-generated with Claude Code on behalf of Federico Mariani.
Co-Authored-By: Claude <[email protected]>
---
.../apache/camel/catalog/docs/git-component.adoc | 15 ++
.../camel-git/src/main/docs/git-component.adoc | 15 ++
.../apache/camel/component/git/GitEndpoint.java | 4 +-
.../git/consumer/AbstractGitConsumer.java | 65 ++++++-
.../component/git/consumer/GitBranchConsumer.java | 3 +-
.../component/git/consumer/GitCommitConsumer.java | 3 +-
.../component/git/consumer/GitTagConsumer.java | 3 +-
.../GitConsumerFailedItemPooledExchangeTest.java | 34 ++++
.../git/consumer/GitConsumerFailedItemTest.java | 216 +++++++++++++++++++++
.../git/consumer/GitConsumerFirstPollTest.java | 129 ++++++++++++
.../git/consumer/GitConsumerOptionsTest.java | 92 +++++++++
.../git/consumer/RecordingExceptionHandler.java | 57 ++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 21 ++
13 files changed, 648 insertions(+), 9 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/git-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/git-component.adoc
index 578506e8aab0..64710f52b5e2 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/git-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/git-component.adoc
@@ -103,6 +103,21 @@ YAML::
----
====
+The consumer polls the repository with the scheduler options of the endpoint,
such as `delay` (500 milliseconds by
+default) and `initialDelay`. Each poll walks the history of the branch, or
lists the tags or the branches, to find the
+items that it has not delivered yet, so raise `delay` for a large repository
that changes seldom.
+
+A commit, tag or branch is marked consumed only when the route processed it
without a failure. When the exchange
+fails, the failure is reported to the exception handler of the consumer, and
the item is delivered again on the next
+poll. It then comes after the items that followed it in the same poll, which
were already delivered: the commits come
+from the newest to the oldest, and the tags and the branches in the order of
their names.
+
+A poll counts only the items that it marked consumed. After a poll whose items
all failed, a `greedy` consumer does
+not poll again at once and `sendEmptyMessageWhenIdle` sends no empty message,
while `backoffIdleThreshold` and
+`backoffMultiplier` count it as an idle poll, so they can slow down the
retries of an item that keeps failing. When an
+item keeps failing, handle the failure in the route, for example with a dead
letter channel or a handled
+`onException`, so that the item is marked consumed.
+
=== Shallow Clone
Use the `depth` option to perform a shallow clone, fetching only a limited
number of commits. This reduces clone time and disk usage for large
repositories when full history is not needed.
diff --git a/components/camel-git/src/main/docs/git-component.adoc
b/components/camel-git/src/main/docs/git-component.adoc
index 578506e8aab0..64710f52b5e2 100644
--- a/components/camel-git/src/main/docs/git-component.adoc
+++ b/components/camel-git/src/main/docs/git-component.adoc
@@ -103,6 +103,21 @@ YAML::
----
====
+The consumer polls the repository with the scheduler options of the endpoint,
such as `delay` (500 milliseconds by
+default) and `initialDelay`. Each poll walks the history of the branch, or
lists the tags or the branches, to find the
+items that it has not delivered yet, so raise `delay` for a large repository
that changes seldom.
+
+A commit, tag or branch is marked consumed only when the route processed it
without a failure. When the exchange
+fails, the failure is reported to the exception handler of the consumer, and
the item is delivered again on the next
+poll. It then comes after the items that followed it in the same poll, which
were already delivered: the commits come
+from the newest to the oldest, and the tags and the branches in the order of
their names.
+
+A poll counts only the items that it marked consumed. After a poll whose items
all failed, a `greedy` consumer does
+not poll again at once and `sendEmptyMessageWhenIdle` sends no empty message,
while `backoffIdleThreshold` and
+`backoffMultiplier` count it as an idle poll, so they can slow down the
retries of an item that keeps failing. When an
+item keeps failing, handle the failure in the route, for example with a dead
letter channel or a handled
+`onException`, so that the item is marked consumed.
+
=== Shallow Clone
Use the `depth` option to perform a shallow clone, fetching only a limited
number of commits. This reduces clone time and disk usage for large
repositories when full history is not needed.
diff --git
a/components/camel-git/src/main/java/org/apache/camel/component/git/GitEndpoint.java
b/components/camel-git/src/main/java/org/apache/camel/component/git/GitEndpoint.java
index 6f2e98bfffb9..48cb56c1df83 100644
---
a/components/camel-git/src/main/java/org/apache/camel/component/git/GitEndpoint.java
+++
b/components/camel-git/src/main/java/org/apache/camel/component/git/GitEndpoint.java
@@ -160,11 +160,13 @@ public class GitEndpoint extends ScheduledPollEndpoint {
throw new IllegalArgumentException("type must be specified for git
consumer");
}
- return switch (type) {
+ Consumer consumer = switch (type) {
case COMMIT -> new GitCommitConsumer(this, processor);
case TAG -> new GitTagConsumer(this, processor);
case BRANCH -> new GitBranchConsumer(this, processor);
};
+ configureConsumer(consumer);
+ return consumer;
}
/**
diff --git
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/AbstractGitConsumer.java
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/AbstractGitConsumer.java
index abd08078cc78..82bb4d2df603 100644
---
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/AbstractGitConsumer.java
+++
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/AbstractGitConsumer.java
@@ -36,6 +36,8 @@ public abstract class AbstractGitConsumer extends
ScheduledBatchPollingConsumer
private final GitEndpoint endpoint;
private Repository repo;
private Git git;
+ // whether the last poll found items to deliver, consumed or not
+ private boolean lastPollFoundItems;
protected AbstractGitConsumer(GitEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -44,9 +46,10 @@ public abstract class AbstractGitConsumer extends
ScheduledBatchPollingConsumer
@Override
protected void doStart() throws Exception {
- super.doStart();
+ // open the repository before super.doStart() starts the scheduler,
which may poll at once (initialDelay=0)
this.repo = getLocalRepository();
this.git = new Git(repo);
+ super.doStart();
}
@Override
@@ -68,10 +71,18 @@ public abstract class AbstractGitConsumer extends
ScheduledBatchPollingConsumer
return git;
}
+ /**
+ * Hands the exchanges of a poll to the route.
+ *
+ * @return the number of items marked consumed. A failed item, or one that
was not handed to the route, is not
+ * counted: a poll whose items all failed does not make a greedy
consumer poll again at once, and the
+ * backoff options count it as an idle poll.
+ */
@Override
public int processBatch(Queue<Object> exchanges) throws Exception {
int total = exchanges.size();
- int answer = total;
+ int consumed = 0;
+ lastPollFoundItems = total > 0;
if (this.maxMessagesPerPoll > 0 && total > this.maxMessagesPerPoll) {
LOG.debug("Limiting to maximum messages to poll {} as there were
{} messages in this poll.",
this.maxMessagesPerPoll, total);
@@ -85,11 +96,55 @@ public abstract class AbstractGitConsumer extends
ScheduledBatchPollingConsumer
exchange.setProperty(ExchangePropertyKey.BATCH_COMPLETE, index ==
total - 1);
this.pendingExchanges = total - index - 1;
Object value = onPreProcessed(exchange);
- getProcessor().process(exchange);
- onProcessed(exchange, value);
+ try {
+ // only an item the route accepted is marked consumed, so a
failed one is delivered again on the
+ // next poll
+ if (processExchange(exchange)) {
+ onProcessed(exchange, value);
+ consumed++;
+ }
+ } finally {
+ releaseExchange(exchange, false);
+ }
+ }
+
+ // release the exchanges that were not handed to the route, when the
consumer stops in the middle of a batch
+ Exchange remaining;
+ while ((remaining = (Exchange) exchanges.poll()) != null) {
+ releaseExchange(remaining, false);
+ }
+
+ return consumed;
+ }
+
+ @Override
+ protected void processEmptyMessage() throws Exception {
+ // a poll whose items all failed counts no item, but it found and
delivered them, so it is not idle
+ if (!lastPollFoundItems) {
+ super.processEmptyMessage();
}
+ }
- return answer;
+ /**
+ * Hands the exchange to the route and reports a failure through the
exception handler of the consumer.
+ * <p/>
+ * A route that fails does not throw from {@code process()}: the failure
is left on the exchange, so it is read
+ * there. The exchange is created with {@code createExchange(false)}, so
it is not reset before its outcome is read.
+ *
+ * @return {@code true} if the route processed the exchange without a
failure
+ */
+ private boolean processExchange(Exchange exchange) {
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ Exception cause = exchange.getException();
+ if (cause != null) {
+ getExceptionHandler().handleException("Error processing exchange",
exchange, cause);
+ return false;
+ }
+ return true;
}
public abstract Object onPreProcessed(Exchange exchange);
diff --git
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java
index 6fb10ce7cedc..25366ddd9092 100644
---
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java
+++
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java
@@ -42,7 +42,8 @@ public class GitBranchConsumer extends AbstractGitConsumer {
List<Ref> call =
getGit().branchList().setListMode(ListMode.ALL).call();
for (Ref ref : call) {
if (!branchesConsumed.contains(ref.getName())) {
- Exchange e = createExchange(true);
+ // released by processBatch once the outcome of the route has
been read
+ Exchange e = createExchange(false);
e.getMessage().setBody(ref.getName());
e.getMessage().setHeader(GitConstants.GIT_BRANCH_NAME,
ref.getName());
e.getMessage().setHeader(GitConstants.GIT_BRANCH_LEAF,
ref.getLeaf().getName());
diff --git
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java
index dd40c7cf10b9..35a7a14995df 100644
---
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java
+++
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java
@@ -59,7 +59,8 @@ public class GitCommitConsumer extends AbstractGitConsumer {
}
for (RevCommit commit : commits) {
if (!commitsConsumed.contains(commit.getId())) {
- Exchange e = createExchange(true);
+ // released by processBatch once the outcome of the route has
been read
+ Exchange e = createExchange(false);
e.getMessage().setBody(commit.getFullMessage());
e.getMessage().setHeader(GitConstants.GIT_COMMIT_ID,
commit.getName());
e.getMessage().setHeader(GitConstants.GIT_COMMIT_AUTHOR_NAME,
commit.getAuthorIdent().getName());
diff --git
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java
index 6f9fe0121700..ba46c50cf657 100644
---
a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java
+++
b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java
@@ -41,7 +41,8 @@ public class GitTagConsumer extends AbstractGitConsumer {
List<Ref> call = getGit().tagList().call();
for (Ref ref : call) {
if (!tagsConsumed.contains(ref.getName())) {
- Exchange e = createExchange(true);
+ // released by processBatch once the outcome of the route has
been read
+ Exchange e = createExchange(false);
e.getMessage().setBody(ref.getName());
e.getMessage().setHeader(GitConstants.GIT_BRANCH_NAME,
ref.getName());
e.getMessage().setHeader(GitConstants.GIT_BRANCH_LEAF,
ref.getLeaf().getName());
diff --git
a/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFailedItemPooledExchangeTest.java
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFailedItemPooledExchangeTest.java
new file mode 100644
index 000000000000..3c4f926fcea3
--- /dev/null
+++
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFailedItemPooledExchangeTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.git.consumer;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+
+/**
+ * The same cases with the pooled exchange factory, which resets an exchange
when its unit of work is done: the consumer
+ * must read the outcome of the exchange before it is released.
+ */
+public class GitConsumerFailedItemPooledExchangeTest extends
GitConsumerFailedItemTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.getCamelContextExtension().setExchangeFactory(new
PooledExchangeFactory());
+ return camelContext;
+ }
+}
diff --git
a/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFailedItemTest.java
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFailedItemTest.java
new file mode 100644
index 000000000000..bd918057fd0c
--- /dev/null
+++
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFailedItemTest.java
@@ -0,0 +1,216 @@
+/*
+ * 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.git.consumer;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.git.GitTestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.support.ScheduledBatchPollingConsumer;
+import org.apache.camel.support.ScheduledPollConsumer;
+import org.eclipse.jgit.api.Git;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+/**
+ * A commit, tag or branch whose exchange failed is reported to the exception
handler and delivered again on the next
+ * poll, and it is marked consumed once the route accepts it. A poll counts
only the items that it marked consumed, so a
+ * failed item does not make a greedy consumer poll again at once, and it
counts as an idle poll for the backoff. The
+ * scheduler is not started: the test polls itself, or runs the scheduled task
itself.
+ */
+public class GitConsumerFailedItemTest extends GitTestSupport {
+
+ private final RecordingExceptionHandler exceptionHandler = new
RecordingExceptionHandler();
+ private final List<String> attempts = new CopyOnWriteArrayList<>();
+ private final AtomicBoolean failing = new AtomicBoolean(true);
+ private volatile int recoverAtAttempt = Integer.MAX_VALUE;
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("recordingExceptionHandler", exceptionHandler);
+ }
+
+ @Test
+ public void failedCommitIsDeliveredAgain() throws Exception {
+ try (Git git = getGitTestRepository()) {
+ commitFile(git);
+ }
+ GitCommitConsumer consumer = (GitCommitConsumer)
context.getRoute("commits").getConsumer();
+
+ assertFailedItemIsDeliveredAgain(consumer::poll, commitMessage);
+ }
+
+ @Test
+ public void failedTagIsDeliveredAgain() throws Exception {
+ try (Git git = getGitTestRepository()) {
+ commitFile(git);
+ git.tag().setName(tagTest).call();
+ }
+ GitTagConsumer consumer = (GitTagConsumer)
context.getRoute("tags").getConsumer();
+
+ assertFailedItemIsDeliveredAgain(consumer::poll, "refs/tags/" +
tagTest);
+ }
+
+ @Test
+ public void failedBranchIsDeliveredAgain() throws Exception {
+ try (Git git = getGitTestRepository()) {
+ commitFile(git);
+ }
+ GitBranchConsumer consumer = (GitBranchConsumer)
context.getRoute("branches").getConsumer();
+
+ assertFailedItemIsDeliveredAgain(consumer::poll, "refs/heads/master");
+ }
+
+ @Test
+ public void greedyConsumerDoesNotPollAgainAfterFailedCommit() throws
Exception {
+ try (Git git = getGitTestRepository()) {
+ commitFile(git);
+ }
+ ScheduledPollConsumer consumer = (ScheduledPollConsumer)
context.getRoute("greedyCommits").getConsumer();
+ // the route recovers at the 10th attempt, so that a consumer that
polls again at once stops there
+ recoverAtAttempt = 10;
+
+ // one run of the scheduler: the commit fails and nothing is consumed,
so the consumer waits for the next run
+ consumer.run();
+ assertEquals(List.of(commitMessage), attempts);
+
+ consumer.run();
+ assertEquals(List.of(commitMessage, commitMessage), attempts);
+ }
+
+ @Test
+ public void greedyConsumerPollsAgainWhileCommitsAreConsumed() throws
Exception {
+ failing.set(false);
+ try (Git git = getGitTestRepository()) {
+ commit(git, "first");
+ commit(git, "second");
+ commit(git, "third");
+ }
+ ScheduledBatchPollingConsumer consumer
+ = (ScheduledBatchPollingConsumer)
context.getRoute("greedyCommits").getConsumer();
+ // not an option of the git endpoint: each poll then consumes one
commit and leaves the older ones
+ consumer.setMaxMessagesPerPoll(1);
+
+ // the commits come from the newest, and the consumer polls again at
once as long as a poll consumed one
+ consumer.run();
+ assertEquals(List.of("third", "second", "first"), attempts);
+ }
+
+ @Test
+ public void failedCommitCountsAsIdlePollForTheBackoff() throws Exception {
+ try (Git git = getGitTestRepository()) {
+ commitFile(git);
+ }
+ ScheduledPollConsumer consumer = (ScheduledPollConsumer)
context.getRoute("backoffCommits").getConsumer();
+
+ // backoffIdleThreshold=1 and backoffMultiplier=3: after a poll that
consumed nothing, 2 runs are skipped
+ for (int run = 0; run < 7; run++) {
+ consumer.run();
+ }
+ assertEquals(List.of(commitMessage, commitMessage, commitMessage),
attempts);
+ }
+
+ @Test
+ public void failedCommitIsNotFollowedByEmptyMessage() throws Exception {
+ try (Git git = getGitTestRepository()) {
+ commitFile(git);
+ }
+ ScheduledPollConsumer consumer = (ScheduledPollConsumer)
context.getRoute("idleCommits").getConsumer();
+
+ // the poll delivered the commit, even if the route failed: it is not
idle
+ consumer.run();
+ assertEquals(List.of(commitMessage), attempts);
+
+ // once the commit is consumed, the next poll finds nothing and sends
an empty message
+ failing.set(false);
+ consumer.run();
+ consumer.run();
+ assertEquals(Arrays.asList(commitMessage, commitMessage, null),
attempts);
+ }
+
+ private void assertFailedItemIsDeliveredAgain(Callable<Integer> poll,
String item) throws Exception {
+ MockEndpoint done = getMockEndpoint("mock:done");
+
+ // the route fails: the failure is reported, and the item is offered
again on the next poll. A poll counts
+ // only the items that it marked consumed.
+ assertEquals(0, poll.call());
+ assertEquals(0, poll.call());
+ assertEquals(List.of(item, item), attempts);
+ assertEquals(List.of(item, item), exceptionHandler.getBodies());
+ assertInstanceOf(IllegalStateException.class,
exceptionHandler.getExceptions().get(0));
+ assertInstanceOf(IllegalStateException.class,
exceptionHandler.getExceptions().get(1));
+ assertEquals(0, done.getReceivedCounter());
+
+ // the route accepts it: the item is marked consumed and not offered
again
+ failing.set(false);
+ assertEquals(1, poll.call());
+ assertEquals(0, poll.call());
+ assertEquals(List.of(item, item, item), attempts);
+ assertEquals(2, exceptionHandler.getBodies().size());
+ assertEquals(1, done.getReceivedCounter());
+ }
+
+ private void commitFile(Git git) throws Exception {
+ new File(getGitDir(), filenameToAdd).createNewFile();
+ git.add().addFilepattern(filenameToAdd).call();
+ git.commit().setMessage(commitMessage).call();
+ }
+
+ private void commit(Git git, String message) throws Exception {
+ git.commit().setMessage(message).setAllowEmpty(true).call();
+ }
+
+ private void attempt(Exchange exchange) {
+ attempts.add(exchange.getMessage().getBody(String.class));
+ if (failing.get() && attempts.size() < recoverAtAttempt) {
+ throw new IllegalStateException("Simulated failure");
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ // force create git repo before routes
+ getTestRepository();
+ final String uri = "git://" + getGitDir().getPath()
+ +
"?startScheduler=false&exceptionHandler=#recordingExceptionHandler&type=";
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(uri +
"commit&branchName=master").routeId("commits").process(e ->
attempt(e)).to("mock:done");
+ from(uri + "tag").routeId("tags").process(e ->
attempt(e)).to("mock:done");
+ from(uri + "branch").routeId("branches").process(e ->
attempt(e)).to("mock:done");
+ from(uri +
"commit&branchName=master&greedy=true").routeId("greedyCommits")
+ .process(e -> attempt(e)).to("mock:done");
+ from(uri +
"commit&branchName=master&backoffIdleThreshold=1&backoffMultiplier=3").routeId("backoffCommits")
+ .process(e -> attempt(e)).to("mock:done");
+ from(uri +
"commit&branchName=master&sendEmptyMessageWhenIdle=true").routeId("idleCommits")
+ .process(e -> attempt(e)).to("mock:done");
+ }
+ };
+ }
+}
diff --git
a/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFirstPollTest.java
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFirstPollTest.java
new file mode 100644
index 000000000000..8818b788ae4b
--- /dev/null
+++
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerFirstPollTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.git.consumer;
+
+import java.io.File;
+import java.util.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.git.GitTestSupport;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.spi.ScheduledPollConsumerScheduler;
+import org.apache.camel.support.service.ServiceSupport;
+import org.eclipse.jgit.api.Git;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The scheduler may run the first poll as soon as the consumer starts it
(initialDelay=0), so the consumer must have
+ * opened the repository before it starts the scheduler. The scheduler of this
test runs the first poll right away, in
+ * the thread that starts the consumer.
+ */
+public class GitConsumerFirstPollTest extends GitTestSupport {
+
+ private final RecordingExceptionHandler exceptionHandler = new
RecordingExceptionHandler();
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("recordingExceptionHandler", exceptionHandler);
+ registry.bind("commitScheduler", new PollOnStartScheduler());
+ registry.bind("tagScheduler", new PollOnStartScheduler());
+ registry.bind("branchScheduler", new PollOnStartScheduler());
+ }
+
+ @Test
+ public void firstPollFindsTheRepository() throws Exception {
+ // the first polls ran while the consumers were started, one route
after the other
+ assertEquals(List.of(), exceptionHandler.getExceptions(), "the first
polls do not fail");
+ List<Object> bodies =
getMockEndpoint("mock:done").getReceivedExchanges().stream()
+ .map(exchange -> exchange.getMessage().getBody())
+ .toList();
+ assertEquals(List.of(commitMessage, "refs/tags/" + tagTest,
"refs/heads/master"), bodies);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ // the repository has a commit, a tag and a branch before the
consumers start
+ try (Git git = getGitTestRepository()) {
+ new File(getGitDir(), filenameToAdd).createNewFile();
+ git.add().addFilepattern(filenameToAdd).call();
+ git.commit().setMessage(commitMessage).call();
+ git.tag().setName(tagTest).call();
+ }
+ final String uri = "git://" + getGitDir().getPath() +
"?exceptionHandler=#recordingExceptionHandler&type=";
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(uri +
"commit&branchName=master&scheduler=#commitScheduler").to("mock:done");
+ from(uri + "tag&scheduler=#tagScheduler").to("mock:done");
+ from(uri +
"branch&scheduler=#branchScheduler").to("mock:done");
+ }
+ };
+ }
+
+ /**
+ * Runs the task once when the scheduler is started.
+ */
+ private static final class PollOnStartScheduler extends ServiceSupport
implements ScheduledPollConsumerScheduler {
+
+ private CamelContext camelContext;
+ private Runnable task;
+ private boolean started;
+
+ @Override
+ public void onInit(Consumer consumer) {
+ // noop
+ }
+
+ @Override
+ public void scheduleTask(Runnable task) {
+ this.task = task;
+ }
+
+ @Override
+ public void unscheduleTask() {
+ task = null;
+ }
+
+ @Override
+ public void startScheduler() {
+ if (!started && task != null) {
+ started = true;
+ task.run();
+ }
+ }
+
+ @Override
+ public boolean isSchedulerStarted() {
+ return started;
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+ }
+}
diff --git
a/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerOptionsTest.java
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerOptionsTest.java
new file mode 100644
index 000000000000..dac3a10c3733
--- /dev/null
+++
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/GitConsumerOptionsTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.git.consumer;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.component.git.GitEndpoint;
+import org.apache.camel.component.git.GitTestSupport;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.support.BridgeExceptionHandlerToErrorHandler;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The scheduler and error handler options of the endpoint are applied to the
consumer of every type.
+ */
+public class GitConsumerOptionsTest extends GitTestSupport {
+
+ private final RecordingExceptionHandler exceptionHandler = new
RecordingExceptionHandler();
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("recordingExceptionHandler", exceptionHandler);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "commit", "tag", "branch" })
+ public void schedulerOptionsReachTheConsumer(String type) throws Exception
{
+ AbstractGitConsumer consumer = createConsumer(type,
"delay=60000&initialDelay=5000&timeUnit=SECONDS"
+ +
"&startScheduler=false&useFixedDelay=false&greedy=true"
+ +
"&backoffMultiplier=4&backoffIdleThreshold=2"
+ +
"&backoffErrorThreshold=3&repeatCount=7"
+ +
"&runLoggingLevel=DEBUG&sendEmptyMessageWhenIdle=true");
+
+ assertEquals(60000, consumer.getDelay());
+ assertEquals(5000, consumer.getInitialDelay());
+ assertEquals(TimeUnit.SECONDS, consumer.getTimeUnit());
+ assertFalse(consumer.isStartScheduler());
+ assertFalse(consumer.isUseFixedDelay());
+ assertTrue(consumer.isGreedy());
+ assertEquals(4, consumer.getBackoffMultiplier());
+ assertEquals(2, consumer.getBackoffIdleThreshold());
+ assertEquals(3, consumer.getBackoffErrorThreshold());
+ assertEquals(7, consumer.getRepeatCount());
+ assertEquals(LoggingLevel.DEBUG, consumer.getRunLoggingLevel());
+ assertTrue(consumer.isSendEmptyMessageWhenIdle());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "commit", "tag", "branch" })
+ public void exceptionHandlerReachesTheConsumer(String type) throws
Exception {
+ AbstractGitConsumer consumer = createConsumer(type,
"exceptionHandler=#recordingExceptionHandler");
+
+ assertSame(exceptionHandler, consumer.getExceptionHandler());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "commit", "tag", "branch" })
+ public void bridgeErrorHandlerReachesTheConsumer(String type) throws
Exception {
+ AbstractGitConsumer consumer = createConsumer(type,
"bridgeErrorHandler=true");
+
+ assertInstanceOf(BridgeExceptionHandlerToErrorHandler.class,
consumer.getExceptionHandler());
+ }
+
+ private AbstractGitConsumer createConsumer(String type, String options)
throws Exception {
+ GitEndpoint endpoint = context.getEndpoint("git://" +
getGitDir().getPath() + "?type=" + type + "&" + options,
+ GitEndpoint.class);
+ return (AbstractGitConsumer) endpoint.createConsumer(exchange -> {
+ });
+ }
+}
diff --git
a/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/RecordingExceptionHandler.java
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/RecordingExceptionHandler.java
new file mode 100644
index 000000000000..132ef420473a
--- /dev/null
+++
b/components/camel-git/src/test/java/org/apache/camel/component/git/consumer/RecordingExceptionHandler.java
@@ -0,0 +1,57 @@
+/*
+ * 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.git.consumer;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.ExceptionHandler;
+
+/**
+ * Records what a consumer reports to its exception handler. The body of the
exchange is read when the failure is
+ * reported, as the consumer may release the exchange afterwards.
+ */
+class RecordingExceptionHandler implements ExceptionHandler {
+
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private final List<Throwable> exceptions = new CopyOnWriteArrayList<>();
+
+ @Override
+ public void handleException(Throwable exception) {
+ handleException(null, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ handleException(message, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange, Throwable
exception) {
+ bodies.add(exchange != null ?
exchange.getMessage().getBody(String.class) : null);
+ exceptions.add(exception);
+ }
+
+ List<String> getBodies() {
+ return bodies;
+ }
+
+ List<Throwable> getExceptions() {
+ return exceptions;
+ }
+}
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 41c97b62f1f8..4de48ed4ae14 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
@@ -7049,3 +7049,24 @@ contains neither the extension nor
`camel-quarkus-observability-services`, which
A project that lists the dependency but does not want the health extension in
the Quarkus export must remove it.
Without the dependency nothing changes: the health extension then only comes
with `--observe` (which
`camel kubernetes export` always sets), as part of
`camel-quarkus-observability-services`.
+
+=== camel-git - the consumer applies the scheduler and error handler options
+
+The `git` consumer (`type=commit`, `tag` and `branch`) now applies the
scheduler options of the endpoint, such as
+`delay`, `initialDelay`, `startScheduler`, `greedy` and the `backoff` options,
and the `bridgeErrorHandler` and
+`exceptionHandler` options. Before, it ignored them and always polled with an
initial delay of 1 second and a delay of
+500 milliseconds. A route that sets one of these options now gets what it
configured: for example, `delay=60000` now
+polls once a minute.
+
+=== camel-git - a failed commit, tag or branch is delivered again on the next
poll
+
+The `git` consumer now marks a commit, tag or branch consumed only when its
exchange completes without a failure. A
+failed exchange (an exception that the route did not handle) is reported to
the exception handler of the consumer,
+which logs it at WARN by default, and the item is delivered again on the next
poll. Before, the item was marked
+consumed whether the route succeeded or not, so a failed item was never
delivered again. The item then comes after
+the items that followed it in the same poll, which were already delivered,
such as the older commits.
+
+An item that fails every time is now routed again at every poll. Handle the
failure in the route, for example with a
+dead letter channel or a handled `onException`, so that the item is marked
consumed. To retry it less often, raise
+`delay`, or set `backoffIdleThreshold` and `backoffMultiplier`: a poll counts
only the items that it marked consumed,
+so a poll whose items all failed is an idle poll, after which a `greedy`
consumer does not poll again at once.
