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 7ae7a2106ea5 CAMEL-24544: camel-kserve / camel-tensorflow-serving -
guard the gRPC channel shutdown in doStop
7ae7a2106ea5 is described below
commit 7ae7a2106ea506f3546d08a84d1554aba7f7bc09
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Sep 1 11:15:44 2026 +0200
CAMEL-24544: camel-kserve / camel-tensorflow-serving - guard the gRPC
channel shutdown in doStop
Both endpoints shut the gRPC channel down unconditionally in doStop
(channel.shutdown()). If doInit() failed before assigning the channel (bad
target,
credentials error), the field is still null and doStop() throws a
NullPointerException that masks the original startup failure. Guard with
if (channel != null) before shutting down, and add a bounded
awaitTermination for
a clean close.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel/component/kserve/KServeEndpoint.java | 9 +++--
.../component/kserve/KServeEndpointDoStopTest.java | 40 +++++++++++++++++++++
.../serving/TensorFlowServingEndpoint.java | 9 +++--
.../TensorFlowServingEndpointDoStopTest.java | 41 ++++++++++++++++++++++
4 files changed, 95 insertions(+), 4 deletions(-)
diff --git
a/components/camel-ai/camel-kserve/src/main/java/org/apache/camel/component/kserve/KServeEndpoint.java
b/components/camel-ai/camel-kserve/src/main/java/org/apache/camel/component/kserve/KServeEndpoint.java
index 74e6af43a9ef..110e230a1ec0 100644
---
a/components/camel-ai/camel-kserve/src/main/java/org/apache/camel/component/kserve/KServeEndpoint.java
+++
b/components/camel-ai/camel-kserve/src/main/java/org/apache/camel/component/kserve/KServeEndpoint.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.kserve;
+import java.util.concurrent.TimeUnit;
+
import inference.GRPCInferenceServiceGrpc;
import io.grpc.ChannelCredentials;
import io.grpc.Grpc;
@@ -73,8 +75,11 @@ public class KServeEndpoint extends DefaultEndpoint {
public void doStop() throws Exception {
super.doStop();
- // Close the channel
- channel.shutdown();
+ // Close the channel if it was created (doInit may have failed before
assigning it)
+ if (channel != null) {
+ channel.shutdown();
+ channel.awaitTermination(5, TimeUnit.SECONDS);
+ }
}
@Override
diff --git
a/components/camel-ai/camel-kserve/src/test/java/org/apache/camel/component/kserve/KServeEndpointDoStopTest.java
b/components/camel-ai/camel-kserve/src/test/java/org/apache/camel/component/kserve/KServeEndpointDoStopTest.java
new file mode 100644
index 000000000000..82ff904cf2f2
--- /dev/null
+++
b/components/camel-ai/camel-kserve/src/test/java/org/apache/camel/component/kserve/KServeEndpointDoStopTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.kserve;
+
+import java.lang.reflect.Field;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+class KServeEndpointDoStopTest {
+
+ @Test
+ void doStopDoesNotThrowWhenChannelWasNeverCreated() throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ KServeEndpoint endpoint =
context.getEndpoint("kserve:infer?target=localhost:8001", KServeEndpoint.class);
+ // Simulate doInit() failing before the gRPC channel was assigned.
+ Field channelField =
KServeEndpoint.class.getDeclaredField("channel");
+ channelField.setAccessible(true);
+ channelField.set(endpoint, null);
+
+ assertDoesNotThrow(endpoint::doStop);
+ }
+ }
+}
diff --git
a/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
b/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
index f0d6d512f81a..96fb7fb731b5 100644
---
a/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
+++
b/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.tensorflow.serving;
+import java.util.concurrent.TimeUnit;
+
import io.grpc.ChannelCredentials;
import io.grpc.Grpc;
import io.grpc.InsecureChannelCredentials;
@@ -71,8 +73,11 @@ public class TensorFlowServingEndpoint extends
DefaultEndpoint {
public void doStop() throws Exception {
super.doStop();
- // Close the channel
- channel.shutdown();
+ // Close the channel if it was created (doInit may have failed before
assigning it)
+ if (channel != null) {
+ channel.shutdown();
+ channel.awaitTermination(5, TimeUnit.SECONDS);
+ }
}
@Override
diff --git
a/components/camel-ai/camel-tensorflow-serving/src/test/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpointDoStopTest.java
b/components/camel-ai/camel-tensorflow-serving/src/test/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpointDoStopTest.java
new file mode 100644
index 000000000000..765665adf39b
--- /dev/null
+++
b/components/camel-ai/camel-tensorflow-serving/src/test/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpointDoStopTest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.tensorflow.serving;
+
+import java.lang.reflect.Field;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+class TensorFlowServingEndpointDoStopTest {
+
+ @Test
+ void doStopDoesNotThrowWhenChannelWasNeverCreated() throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ TensorFlowServingEndpoint endpoint = context.getEndpoint(
+ "tensorflow-serving:predict?target=localhost:8500",
TensorFlowServingEndpoint.class);
+ // Simulate doInit() failing before the gRPC channel was assigned.
+ Field channelField =
TensorFlowServingEndpoint.class.getDeclaredField("channel");
+ channelField.setAccessible(true);
+ channelField.set(endpoint, null);
+
+ assertDoesNotThrow(endpoint::doStop);
+ }
+ }
+}
