This is an automated email from the ASF dual-hosted git repository.
Croway 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 50d17f9c2614 CAMEL-25838: camel-iggy - a consumer client created after
a failed poll joins the consumer group (#28259)
50d17f9c2614 is described below
commit 50d17f9c26140cf55deae517310b0a310e3e6c5b
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 22 18:46:47 2026 +0530
CAMEL-25838: camel-iggy - a consumer client created after a failed poll
joins the consumer group (#28259)
* CAMEL-25838: camel-iggy - a consumer client created after a failed poll
joins the consumer group
The consumer joined its consumer group once, with the client it borrowed
when it started. Since CAMEL-25289 the client of a failed poll is
discarded and the pool creates a new client, which only logs in. The Iggy
server keeps the group membership per connection and answers a poll of a
client that is not a member with ConsumerGroupMemberNotFound, so after one
failed poll (a dropped connection, a server restart) every poll failed and
nothing was consumed until the route was restarted.
The consumer now remembers which clients joined the group and makes a
client join before its first poll.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25838: camel-iggy - use AssertJ in IggyConsumerRejoinGroupTest, as
the other unit tests of the module
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/iggy/IggyConsumer.java | 31 +++++
.../apache/camel/component/iggy/IggyEndpoint.java | 8 ++
.../camel/component/iggy/IggyFetchRecords.java | 3 +
.../iggy/IggyConsumerRejoinGroupTest.java | 150 +++++++++++++++++++++
4 files changed, 192 insertions(+)
diff --git
a/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyConsumer.java
b/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyConsumer.java
index e331f9625a84..d0326e90eb33 100644
---
a/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyConsumer.java
+++
b/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyConsumer.java
@@ -17,7 +17,10 @@
package org.apache.camel.component.iggy;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@@ -36,6 +39,9 @@ public class IggyConsumer extends DefaultConsumer {
private IggyClientConnectionPool iggyClientConnectionPool;
private ExecutorService executor;
private final List<IggyFetchRecords> tasks = new ArrayList<>();
+ // the pooled clients that joined the consumer group (the membership
belongs to the connection of a client)
+ private final Set<IggyBaseClient> groupMembers
+ = Collections.synchronizedSet(Collections.newSetFromMap(new
IdentityHashMap<>()));
public IggyConsumer(IggyEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -62,6 +68,7 @@ public class IggyConsumer extends DefaultConsumer {
try {
endpoint.initializeTopic(client);
endpoint.initializeConsumerGroup(client);
+ groupMembers.add(client);
} finally {
iggyClientConnectionPool.returnClient(client);
}
@@ -112,6 +119,7 @@ public class IggyConsumer extends DefaultConsumer {
}
}
tasks.clear();
+ groupMembers.clear();
executor = null;
if (iggyClientConnectionPool != null) {
iggyClientConnectionPool.close();
@@ -120,4 +128,27 @@ public class IggyConsumer extends DefaultConsumer {
super.doStop();
}
+
+ /**
+ * Makes the client a member of the consumer group unless it joined
already. A client the pool created after a
+ * failed poll is a new connection, which the server does not know as a
member: its polls would fail with
+ * ConsumerGroupMemberNotFound.
+ */
+ void joinConsumerGroupIfNeeded(IggyBaseClient client) {
+ if (!groupMembers.contains(client)) {
+ endpoint.joinConsumerGroup(client);
+ groupMembers.add(client);
+ LOG.info("Joined the consumer group {} of stream {} and topic {}
again with a new client"
+ + " (the client of a failed poll was discarded)",
+ endpoint.getConfiguration().getConsumerGroupName(),
endpoint.getConfiguration().getStreamName(),
+ endpoint.getTopicName());
+ }
+ }
+
+ /**
+ * Forgets a client that is discarded.
+ */
+ void clientDiscarded(IggyBaseClient client) {
+ groupMembers.remove(client);
+ }
}
diff --git
a/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyEndpoint.java
b/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyEndpoint.java
index a8a68f9b6b64..b9d340db7061 100644
---
a/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyEndpoint.java
+++
b/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyEndpoint.java
@@ -178,6 +178,14 @@ public class IggyEndpoint extends DefaultEndpoint
implements HeaderFilterStrateg
return consumerGroupDetails;
});
+ joinConsumerGroup(client);
+ }
+
+ /**
+ * Makes the client a member of the consumer group. The membership belongs
to the connection of the client, so each
+ * client the consumer polls with must join.
+ */
+ public void joinConsumerGroup(IggyBaseClient client) {
client.consumerGroups().joinConsumerGroup(
StreamId.of(getConfiguration().getStreamName()),
TopicId.of(getTopicName()),
diff --git
a/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyFetchRecords.java
b/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyFetchRecords.java
index 4192edcf2712..5dd67287e722 100644
---
a/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyFetchRecords.java
+++
b/components/camel-iggy/src/main/java/org/apache/camel/component/iggy/IggyFetchRecords.java
@@ -97,6 +97,8 @@ public class IggyFetchRecords implements Runnable {
IggyBaseClient client = iggyClientConnectionPool.borrowObject();
boolean polled = false;
try {
+ // a client created after a failed poll is not a member of the
group yet
+ iggyConsumer.joinConsumerGroupIfNeeded(client);
polledMessages = poll(client, streamId, topicId, consumerId);
polled = true;
} finally {
@@ -104,6 +106,7 @@ public class IggyFetchRecords implements Runnable {
iggyClientConnectionPool.returnClient(client);
} else {
// the client of a failed poll may be broken (connection
lost): do not hand it out again
+ iggyConsumer.clientDiscarded(client);
iggyClientConnectionPool.invalidateClient(client);
}
}
diff --git
a/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyConsumerRejoinGroupTest.java
b/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyConsumerRejoinGroupTest.java
new file mode 100644
index 000000000000..c5c28f0d24fc
--- /dev/null
+++
b/components/camel-iggy/src/test/java/org/apache/camel/component/iggy/IggyConsumerRejoinGroupTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.iggy;
+
+import java.io.Closeable;
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.component.iggy.client.IggyClientFactory;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.iggy.client.blocking.ConsumerGroupsClient;
+import org.apache.iggy.client.blocking.IggyBaseClient;
+import org.apache.iggy.consumergroup.ConsumerGroupDetails;
+import org.apache.iggy.exception.IggyErrorCode;
+import org.apache.iggy.exception.IggyResourceNotFoundException;
+import org.apache.iggy.identifier.ConsumerId;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.PolledMessages;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+/**
+ * The consumer group membership belongs to the connection of a client: the
Iggy server answers a poll of a client that
+ * did not join the group with ConsumerGroupMemberNotFound. After a failed
poll the consumer discards its client and the
+ * pool creates a new one, which must join the group before it polls. The Iggy
clients are mocks that behave like the
+ * server in this respect; no Iggy server is needed.
+ */
+class IggyConsumerRejoinGroupTest {
+
+ private static final String URI
+ =
"iggy:topic?streamName=stream&autoCreateStream=false&autoCreateTopic=false&consumerGroupName=group";
+
+ private final List<String> received = new CopyOnWriteArrayList<>();
+ private final List<IggyBaseClient> clients = new CopyOnWriteArrayList<>();
+
+ // how the first client fails its first poll
+ private boolean firstClientReconnectedOnItsOwn;
+
+ @Test
+ void testNewClientJoinsTheGroupAfterAFailedPoll() throws Exception {
+ consumeAfterTheFirstClientFailed();
+ }
+
+ @Test
+ void testNewClientJoinsTheGroupAfterTheClientReconnectedOnItsOwn() throws
Exception {
+ // the iggy 0.9.0 TCP client replaces a lost connection and logs in
again by itself (a channel pool of one
+ // connection that replays the login), but does not join the consumer
group again: the server answers its next
+ // poll with ConsumerGroupMemberNotFound (5006), although the consumer
saw it join
+ firstClientReconnectedOnItsOwn = true;
+ consumeAfterTheFirstClientFailed();
+ }
+
+ private void consumeAfterTheFirstClientFailed() throws Exception {
+ try (MockedConstruction<IggyClientFactory> factories =
mockConstruction(IggyClientFactory.class,
+ withSettings().defaultAnswer(CALLS_REAL_METHODS),
+ (factory, context) -> doAnswer(invocation ->
newClient()).when(factory).create());
+ CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ Consumer consumer =
context.getEndpoint(URI).createConsumer(exchange -> {
+ received.add(exchange.getIn().getBody(String.class));
+ });
+ consumer.start();
+
+ // the first client loses its connection at its first poll; the
next client must consume
+ await().atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(received).containsExactly("after the failed poll"));
+
+ consumer.stop();
+ }
+ }
+
+ private IggyBaseClient newClient() throws Exception {
+ boolean first = clients.isEmpty();
+ AtomicBoolean member = new AtomicBoolean();
+ AtomicBoolean delivered = new AtomicBoolean();
+ IggyBaseClient client = mock(IggyBaseClient.class,
+
withSettings().extraInterfaces(Closeable.class).defaultAnswer(RETURNS_DEEP_STUBS));
+ clients.add(client);
+
+ // the group exists; joining makes this client (its connection) a
member
+ ConsumerGroupDetails group = mock(ConsumerGroupDetails.class);
+ ConsumerGroupsClient groups = client.consumerGroups();
+ when(groups.getConsumerGroup(any(StreamId.class), any(TopicId.class),
any(ConsumerId.class)))
+ .thenReturn(Optional.of(group));
+ doAnswer(invocation -> {
+ member.set(true);
+ return null;
+ }).when(groups).joinConsumerGroup(any(StreamId.class),
any(TopicId.class), any(ConsumerId.class));
+
+ when(client.messages().pollMessages(any(StreamId.class),
any(TopicId.class), any(), any(), any(), any(),
+ anyBoolean())).thenAnswer(invocation -> {
+ if (first && firstClientReconnectedOnItsOwn) {
+ throw memberNotFound();
+ }
+ if (first) {
+ // any failure of the poll discards the client (iggy
throws IggyConnectionException and others)
+ throw new IllegalStateException("Connection reset");
+ }
+ if (!member.get()) {
+ throw memberNotFound();
+ }
+ if (delivered.compareAndSet(false, true)) {
+ return new PolledMessages(
+ 0L, BigInteger.ZERO, 1L,
List.of(Message.of("after the failed poll")));
+ }
+ return new PolledMessages(0L, BigInteger.ZERO, 0L,
List.of());
+ });
+ return client;
+ }
+
+ private static IggyResourceNotFoundException memberNotFound() {
+ return new IggyResourceNotFoundException(
+ IggyErrorCode.CONSUMER_GROUP_MEMBER_NOT_FOUND, 5006,
"consumer_group_member_not_found",
+ Optional.empty(), Optional.empty());
+ }
+}
