gnodet-bot commented on code in PR #29489: URL: https://github.com/apache/camel/pull/29489#discussion_r4514887995
########## components/camel-azure/camel-azure-iothub/src/main/java/org/apache/camel/component/azure/iothub/operations/TwinDeliveryStrategy.java: ########## @@ -0,0 +1,120 @@ +/* + * 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.azure.iothub.operations; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import com.microsoft.azure.sdk.iot.device.DeviceClient; +import com.microsoft.azure.sdk.iot.device.twin.Twin; +import com.microsoft.azure.sdk.iot.device.twin.TwinCollection; +import org.apache.camel.component.azure.iothub.IoTHubDeliveryMode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Updates Azure IoT Hub device twin reported properties from a {@code Map<String, Object>} body. + * + * <p> + * The body must be a {@link Map} of reported properties (any payload decoding, e.g. Protobuf/CloudEvent to Map, is the + * responsibility of a preceding processor in the route). Updates are delta-based: only changed properties are sent. + * </p> + */ +public class TwinDeliveryStrategy implements DeliveryStrategy { + + private static final Logger LOG = LoggerFactory.getLogger(TwinDeliveryStrategy.class); + + @Override + public void execute(DeviceClient client, String deviceId, DeliveryContext context) throws Exception { + Map<String, Object> twinProperties = context.twinProperties(); + if (twinProperties == null) { + throw new IllegalArgumentException( + "Twin delivery expects twin properties (a Map) to be provided either as the message body " + + "or via the CamelAzureIoTHubTwinProperties header"); + } + + Map<String, Object> propertyMap = clean(twinProperties); + + client.subscribeToDesiredProperties((twinUpdate, ctx) -> { Review Comment: 🚨 **`subscribeToDesiredProperties` called per message**: This method is a SDK subscription call — it should be invoked once when the client is initialized, not on every `execute()` call. Currently, every twin delivery re-registers a no-op callback and then fetches the twin. Move the subscription to `DeviceClientPool.createClient()` (once per pooled client lifetime), or at minimum gate it with a flag on the `PooledClient`. ########## components/camel-azure/camel-azure-iothub/src/main/java/org/apache/camel/component/azure/iothub/IoTHubEndpoint.java: ########## @@ -0,0 +1,62 @@ +/* + * 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.azure.iothub; + +import org.apache.camel.Category; +import org.apache.camel.Component; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.spi.UriEndpoint; +import org.apache.camel.spi.UriParam; +import org.apache.camel.support.DefaultEndpoint; + +/** + * Send device telemetry and device-twin updates to Azure IoT Hub. + */ +@UriEndpoint(firstVersion = "4.16.0", scheme = "azure-iothub", title = "Azure IoT Hub", Review Comment: 🚨 **Wrong `firstVersion`**: This component is being introduced in `4.24.0-SNAPSHOT`, but `firstVersion` is set to `4.16.0` — a version released long ago. Update this to the actual release version in which the component ships. The generated JSON will also need to be regenerated after the fix. ########## components/camel-azure/camel-azure-iothub/src/main/java/org/apache/camel/component/azure/iothub/client/DeviceClientPool.java: ########## @@ -0,0 +1,202 @@ +/* + * 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.azure.iothub.client; + +import java.time.Duration; +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.LongSupplier; + +import com.microsoft.azure.sdk.iot.device.DeviceClient; +import com.microsoft.azure.sdk.iot.device.IotHubClientProtocol; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Thread-safe pool of Azure IoT Hub {@link DeviceClient} instances, keyed by device id. + * + * <p> + * Clients are created lazily, reused across messages, and evicted after an idle timeout. The Azure device SDK does not + * expose connection state, so broken clients are removed by the caller on send failure (see the send operations). + * </p> + */ +public class DeviceClientPool { + + private static final Logger LOG = LoggerFactory.getLogger(DeviceClientPool.class); + + @FunctionalInterface + public interface DeviceClientFactory { + DeviceClient create(String connectionString, IotHubClientProtocol protocol) throws Exception; + } + + private static final class PooledClient { + final DeviceClient client; + volatile long lastUsed; + + PooledClient(DeviceClient client, long nowMillis) { + this.client = client; + this.lastUsed = nowMillis; + } + + void touch(long nowMillis) { + this.lastUsed = nowMillis; + } + } + + private final Duration idleTimeout; + private final Duration evictionInterval; + private final Map<String, PooledClient> pool = new ConcurrentHashMap<>(); + private final AtomicBoolean shuttingDown = new AtomicBoolean(false); + + private DeviceClientFactory deviceClientFactory = DeviceClient::new; + private LongSupplier nowMillisSupplier = System::currentTimeMillis; + private ScheduledExecutorService evictionExecutor; + + public DeviceClientPool(Duration idleTimeout, Duration evictionInterval) { + this.idleTimeout = isPositive(idleTimeout) ? idleTimeout : Duration.ofMinutes(5); + this.evictionInterval = isPositive(evictionInterval) ? evictionInterval : Duration.ofSeconds(30); + } + + private static boolean isPositive(Duration d) { + return d != null && !d.isZero() && !d.isNegative(); + } + + public void start() { + evictionExecutor = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "camel-azure-iothub-device-client-evictor"); + t.setDaemon(true); + return t; + }); + evictionExecutor.scheduleAtFixedRate(this::evictIdleClients, + evictionInterval.toMillis(), evictionInterval.toMillis(), TimeUnit.MILLISECONDS); + LOG.info("DeviceClient eviction enabled. idleTimeout={}s, interval={}s", + idleTimeout.getSeconds(), evictionInterval.getSeconds()); + } + + public DeviceClient getOrCreate(String deviceId, String connectionString, IotHubClientProtocol protocol) { + if (Thread.currentThread().isInterrupted() || shuttingDown.get()) { + throw new IllegalStateException("Skipping client creation due to shutdown: " + deviceId); + } + final long now = nowMillisSupplier.getAsLong(); + + PooledClient pooled = pool.compute(deviceId, (id, current) -> { + if (current != null) { + current.touch(now); + return current; + } + DeviceClient client = createClient(id, connectionString, protocol); + return client == null ? null : new PooledClient(client, now); + }); + + if (pooled == null) { + throw new IllegalStateException("Client creation interrupted due to shutdown: " + deviceId); + } + pooled.touch(now); + return pooled.client; + } + + private DeviceClient createClient(String deviceId, String connectionString, IotHubClientProtocol protocol) { + try { + DeviceClient client = deviceClientFactory.create(connectionString, protocol); + client.open(true); + LOG.info("Created new DeviceClient for device: {}", deviceId); + return client; + } catch (Exception e) { + if (isShutdownInterruption(e)) { + Thread.currentThread().interrupt(); + LOG.warn("DeviceClient creation interrupted due to shutdown for device: {}", deviceId); + return null; + } + throw new RuntimeException("DeviceClient creation failed for device: " + deviceId, e); + } + } + + public void remove(String deviceId) { + PooledClient pooled = pool.remove(deviceId); + if (pooled != null) { + closeQuietly(deviceId, pooled.client, "explicit removal"); + } + } + + private void evictIdleClients() { + if (shuttingDown.get()) { + return; + } + long now = nowMillisSupplier.getAsLong(); + Iterator<Map.Entry<String, PooledClient>> it = pool.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry<String, PooledClient> entry = it.next(); + if ((now - entry.getValue().lastUsed) > idleTimeout.toMillis() + && pool.remove(entry.getKey(), entry.getValue())) { Review Comment: ⚠️ **Use-after-eviction race**: The evictor calls `pool.remove(entry.getKey(), entry.getValue())` and then `close()` on the client. Meanwhile a caller may already hold the `DeviceClient` reference returned by `getOrCreate` and be mid-use. `touch()` on line 1277 (after `compute`) does not prevent this — the evictor may have already captured the entry before the touch. Consider a reference-counting lease (increment on `getOrCreate`, decrement when done) so the evictor skips clients with active leases, or document this as a known limitation with a note that callers should handle `IllegalStateException` from a closed client. ########## components/camel-azure/camel-azure-iothub/src/main/docs/azure-iothub-component.adoc: ########## @@ -0,0 +1,127 @@ += Azure IoT Hub Component +:doctitle: Azure IoT Hub +:shortname: azure-iothub +:artifactid: camel-azure-iothub +:description: Send device telemetry and device-twin updates to Azure IoT Hub. +:since: 4.16 +:supportlevel: Stable +:tabs-sync-option: +:component-header: Only producer is supported +//Manually maintained attributes +:group: Azure + +*Since Camel {since}* + +*{component-header}* + +The Azure IoT Hub component provides the capability to send device-to-cloud +telemetry and to update device-twin reported properties on +https://azure.microsoft.com/services/iot-hub/[Azure IoT Hub], using the +Azure IoT device SDK. Device connections are pooled and reused per device id. + +*Prerequisites* + +You must have a valid Microsoft Azure IoT Hub and a per-device connection string. +More information is available at the +https://learn.microsoft.com/azure/iot-hub/[Azure IoT Hub Documentation]. + +Maven users will need to add the following dependency to their `pom.xml` +for this component: + +[source,xml] +------------------------------------------------------------ +<dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-azure-iothub</artifactId> + <version>x.x.x</version> + <!-- use the same version as your Camel core version --> +</dependency> +------------------------------------------------------------ + +== URI Format + +[source] +------------------------------------------------------------ +azure-iothub:deviceId[?options] +------------------------------------------------------------ + +// component-configure options: START + +// component-configure options: END + +// component options: START +include::partial$component-configure-options.adoc[] +include::partial$component-endpoint-options.adoc[] +include::partial$component-endpoint-headers.adoc[] Review Comment: ⚠️ **Duplicate include**: `component-endpoint-headers.adoc` is included here (inside `// component options: START/END`) AND again at line 62 (standalone, after `// endpoint options: END`). The first occurrence (this one) is also in the wrong block — headers belong after endpoint options, not inside the component options block. Remove this include; keep only the one at line 62. ########## components/camel-azure/camel-azure-iothub/src/main/java/org/apache/camel/component/azure/iothub/client/DeviceClientPool.java: ########## @@ -0,0 +1,202 @@ +/* + * 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.azure.iothub.client; + +import java.time.Duration; +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.LongSupplier; + +import com.microsoft.azure.sdk.iot.device.DeviceClient; +import com.microsoft.azure.sdk.iot.device.IotHubClientProtocol; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Thread-safe pool of Azure IoT Hub {@link DeviceClient} instances, keyed by device id. + * + * <p> + * Clients are created lazily, reused across messages, and evicted after an idle timeout. The Azure device SDK does not + * expose connection state, so broken clients are removed by the caller on send failure (see the send operations). + * </p> + */ +public class DeviceClientPool { + + private static final Logger LOG = LoggerFactory.getLogger(DeviceClientPool.class); + + @FunctionalInterface + public interface DeviceClientFactory { + DeviceClient create(String connectionString, IotHubClientProtocol protocol) throws Exception; + } + + private static final class PooledClient { + final DeviceClient client; + volatile long lastUsed; + + PooledClient(DeviceClient client, long nowMillis) { + this.client = client; + this.lastUsed = nowMillis; + } + + void touch(long nowMillis) { + this.lastUsed = nowMillis; + } + } + + private final Duration idleTimeout; + private final Duration evictionInterval; + private final Map<String, PooledClient> pool = new ConcurrentHashMap<>(); + private final AtomicBoolean shuttingDown = new AtomicBoolean(false); + + private DeviceClientFactory deviceClientFactory = DeviceClient::new; + private LongSupplier nowMillisSupplier = System::currentTimeMillis; + private ScheduledExecutorService evictionExecutor; + + public DeviceClientPool(Duration idleTimeout, Duration evictionInterval) { + this.idleTimeout = isPositive(idleTimeout) ? idleTimeout : Duration.ofMinutes(5); + this.evictionInterval = isPositive(evictionInterval) ? evictionInterval : Duration.ofSeconds(30); + } + + private static boolean isPositive(Duration d) { + return d != null && !d.isZero() && !d.isNegative(); + } + + public void start() { + evictionExecutor = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "camel-azure-iothub-device-client-evictor"); + t.setDaemon(true); + return t; + }); + evictionExecutor.scheduleAtFixedRate(this::evictIdleClients, + evictionInterval.toMillis(), evictionInterval.toMillis(), TimeUnit.MILLISECONDS); + LOG.info("DeviceClient eviction enabled. idleTimeout={}s, interval={}s", + idleTimeout.getSeconds(), evictionInterval.getSeconds()); + } + + public DeviceClient getOrCreate(String deviceId, String connectionString, IotHubClientProtocol protocol) { + if (Thread.currentThread().isInterrupted() || shuttingDown.get()) { + throw new IllegalStateException("Skipping client creation due to shutdown: " + deviceId); + } + final long now = nowMillisSupplier.getAsLong(); + + PooledClient pooled = pool.compute(deviceId, (id, current) -> { Review Comment: 🚨 **Blocking I/O inside `ConcurrentHashMap.compute`**: `createClient()` calls `client.open(true)`, which is a blocking network operation. `ConcurrentHashMap.compute` holds the per-bucket lock for the entire lambda duration, so any thread accessing a key in the same hash bucket will stall until the connection completes. Under concurrent load with multiple devices, this can cause cascading delays. Suggested fix: use `putIfAbsent` or a two-phase approach — allocate the `PooledClient` placeholder outside the lock, perform `open()`, then install it, failing gracefully if another thread raced. ########## components/camel-azure/camel-azure-iothub/src/test/java/org/apache/camel/component/azure/iothub/client/DeviceClientPoolTest.java: ########## @@ -0,0 +1,124 @@ +/* + * 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.azure.iothub.client; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import com.microsoft.azure.sdk.iot.device.DeviceClient; +import com.microsoft.azure.sdk.iot.device.IotHubClientProtocol; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +class DeviceClientPoolTest { + + private static final String CS = "HostName=h;DeviceId=d;SharedAccessKey=k"; + + @Test + void testGetOrCreateCreatesOnceAndReuses() throws Exception { + AtomicInteger created = new AtomicInteger(); + DeviceClient client = mock(DeviceClient.class); + + DeviceClientPool pool = new DeviceClientPool(Duration.ofMinutes(5), Duration.ofSeconds(30)); + pool.setDeviceClientFactory((cs, protocol) -> { + created.incrementAndGet(); + return client; + }); + + DeviceClient first = pool.getOrCreate("dev1", CS, IotHubClientProtocol.MQTT); + DeviceClient second = pool.getOrCreate("dev1", CS, IotHubClientProtocol.MQTT); + + assertSame(client, first); + assertSame(first, second); + assertEquals(1, created.get(), "client should be created once and reused"); + assertEquals(1, pool.getActiveClientCount()); + verify(client, times(1)).open(true); + } + + @Test + void testRemoveClosesClient() throws Exception { + DeviceClient client = mock(DeviceClient.class); + DeviceClientPool pool = new DeviceClientPool(Duration.ofMinutes(5), Duration.ofSeconds(30)); + pool.setDeviceClientFactory((cs, protocol) -> client); + + pool.getOrCreate("dev1", CS, IotHubClientProtocol.MQTT); + pool.remove("dev1"); + + assertEquals(0, pool.getActiveClientCount()); + verify(client, times(1)).close(); + } + + @Test + void testShutdownClosesAllAndBlocksCreation() throws Exception { + DeviceClient client = mock(DeviceClient.class); + DeviceClientPool pool = new DeviceClientPool(Duration.ofMinutes(5), Duration.ofSeconds(30)); + pool.setDeviceClientFactory((cs, protocol) -> client); + + pool.getOrCreate("dev1", CS, IotHubClientProtocol.MQTT); + pool.shutdown(); + + assertEquals(0, pool.getActiveClientCount()); + verify(client, times(1)).close(); + assertThrows(IllegalStateException.class, () -> pool.getOrCreate("dev2", CS, IotHubClientProtocol.MQTT)); + } + + @Test + void testIdleClientsAreEvicted() throws Exception { + DeviceClient client = mock(DeviceClient.class); + AtomicLong clock = new AtomicLong(0); + + DeviceClientPool pool = new DeviceClientPool(Duration.ofMillis(10), Duration.ofMillis(20)); + pool.setDeviceClientFactory((cs, protocol) -> client); + pool.setNowMillisSupplier(clock::get); + pool.start(); + try { + pool.getOrCreate("dev1", CS, IotHubClientProtocol.MQTT); + assertEquals(1, pool.getActiveClientCount()); + + // advance well beyond the idle timeout so the scheduled evictor removes it + clock.set(10_000); + + long deadline = System.currentTimeMillis() + 2000; + while (pool.getActiveClientCount() != 0 && System.currentTimeMillis() < deadline) { + Thread.sleep(10); Review Comment: 💡 **`Thread.sleep` in test**: This polling loop with `Thread.sleep(10)` and a 2-second deadline is timing-sensitive. Consider using `Awaitility` for a more robust, readable assertion: ```java await().atMost(2, SECONDS).untilAsserted(() -> assertEquals(0, pool.getActiveClientCount(), "idle client should have been evicted")); ``` ########## components/camel-azure/camel-azure-iothub/src/main/java/org/apache/camel/component/azure/iothub/IoTHubConstants.java: ########## @@ -0,0 +1,52 @@ +/* + * 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.azure.iothub; + +import org.apache.camel.spi.Metadata; + +/** + * Constants (message headers) used by the Azure IoT Hub component. + */ +public final class IoTHubConstants { + + private static final String HEADER_PREFIX = "CamelAzureIoTHub"; + + @Metadata(label = "producer", + description = "The target device id. Overrides the device id configured on the endpoint.", + javaType = "String") + public static final String DEVICE_ID = HEADER_PREFIX + "DeviceId"; + + @Metadata(label = "producer", Review Comment: ⚠️ **Missing `secret = true`**: The `CONNECTION_STRING` header carries a device-level credential (IoT Hub connection string), but the `@Metadata` annotation is missing `secret = true`. The generated JSON confirms `"secret": false` for this header. Add `secret = true` to be consistent with the endpoint configuration field, which correctly declares it secret. ```suggestion @Metadata(label = "producer", description = "The per-device IoT Hub connection string. Overrides the connection string " + "configured on the endpoint.", javaType = "String", secret = true) ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
