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 1ba98b8108d6 CAMEL-23996: Fix SingleNodeKafkaResumeStrategy
correctness bugs
1ba98b8108d6 is described below
commit 1ba98b8108d6c132921c4a7c33d922c5eff94f77
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 09:03:47 2026 +0200
CAMEL-23996: Fix SingleNodeKafkaResumeStrategy correctness bugs
Fix three bugs in SingleNodeKafkaResumeStrategy:
- Double subscribe replaces rebalance listener: refresh() called
subscribe(consumer) which registers a ConsumerRebalanceListener, then
immediately called consumer.subscribe() again without the listener,
replacing the subscription. Removed the redundant second subscribe.
- getAdapter() never waits for initialization: the condition adapter==null
was never true because setAdapter() is always called before loadCache().
Changed to check initLatch != null so callers block until cache loading
completes.
- stop() throws IllegalMonitorStateException: writeLock.tryLock() can
return false, but the finally block always called unlock(). Added
isHeldByCurrentThread() guard before unlocking.
Closes #24692
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../kafka/SingleNodeKafkaResumeStrategy.java | 8 +-
.../kafka/SingleNodeKafkaResumeStrategyTest.java | 104 +++++++++++++++++++++
2 files changed, 108 insertions(+), 4 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
index 32850784eaab..7bd0ad00875b 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
@@ -209,8 +209,6 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
subscribe(consumer);
LOG.debug("Loading records from topic {}",
resumeStrategyConfiguration.getTopic());
-
consumer.subscribe(Collections.singletonList(resumeStrategyConfiguration.getTopic()));
-
poll(consumer, latch);
} catch (WakeupException e) {
LOG.info("Kafka consumer was interrupted during a blocking call");
@@ -358,7 +356,7 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
@Override
public ResumeAdapter getAdapter() {
- if (adapter == null) {
+ if (initLatch != null) {
waitForInitialization();
}
@@ -405,7 +403,9 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
} catch (Exception e) {
LOG.warn("Error closing the Kafka producer: {} (this error will be
ignored)", e.getMessage(), e);
} finally {
- writeLock.unlock();
+ if (writeLock.isHeldByCurrentThread()) {
+ writeLock.unlock();
+ }
}
try {
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategyTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategyTest.java
new file mode 100644
index 000000000000..521152ace7c9
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategyTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.processor.resume.kafka;
+
+import java.lang.reflect.Field;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.camel.resume.ResumeAdapter;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Unit tests for {@link SingleNodeKafkaResumeStrategy}.
+ */
+public class SingleNodeKafkaResumeStrategyTest {
+
+ @Test
+ void testStopDoesNotThrowWhenLockNotAcquired() throws Exception {
+ SingleNodeKafkaResumeStrategy strategy = new
SingleNodeKafkaResumeStrategy();
+
+ ReentrantLock lock = getField(strategy, "writeLock");
+ // Acquire the lock from another thread so tryLock fails in stop()
+ Thread holder = new Thread(lock::lock);
+ holder.start();
+ holder.join();
+
+ // stop() should not throw IllegalMonitorStateException
+ assertDoesNotThrow(strategy::stop);
+
+ // Release the lock from the holder thread
+ Thread releaser = new Thread(lock::unlock);
+ releaser.start();
+ releaser.join();
+ }
+
+ @Test
+ void testGetAdapterReturnsAdapterWithoutLoadCache() {
+ SingleNodeKafkaResumeStrategy strategy = new
SingleNodeKafkaResumeStrategy();
+
+ ResumeAdapter adapter = new TestAdapter();
+ strategy.setAdapter(adapter);
+
+ assertSame(adapter, strategy.getAdapter());
+ }
+
+ @Test
+ void testGetAdapterWaitsForInitializationWhenLatchIsSet() throws Exception
{
+ Properties props = new Properties();
+ props.put("bootstrap.servers", "localhost:9092");
+ KafkaResumeStrategyConfiguration config
+ = new KafkaResumeStrategyConfigurationBuilder(props, props)
+ .withTopic("test-topic")
+ .build();
+
+ SingleNodeKafkaResumeStrategy strategy = new
SingleNodeKafkaResumeStrategy(config);
+
+ ResumeAdapter adapter = new TestAdapter();
+ strategy.setAdapter(adapter);
+
+ CountDownLatch latch = new CountDownLatch(1);
+ setField(strategy, "initLatch", latch);
+
+ // Count down so waitForInitialization() completes immediately
+ latch.countDown();
+
+ assertSame(adapter, strategy.getAdapter());
+ }
+
+ static class TestAdapter implements ResumeAdapter {
+ @Override
+ public void resume() {
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T getField(Object target, String fieldName) throws
Exception {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (T) field.get(target);
+ }
+
+ private static void setField(Object target, String fieldName, Object
value) throws Exception {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+}
