From f9c84264615d6437de519963fd4e920506385342 Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Wed, 12 Nov 2025 16:03:57 +0200 Subject: [PATCH 1/5] snmp context run bootstrap async with process-level retries --- .../src/main/resources/thingsboard.yml | 2 + .../transport/snmp/SnmpTransportContext.java | 38 ++++++++++++++++--- 2 files changed, 34 insertions(+), 6 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 7da26fdc81..81fe17c2b6 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1295,6 +1295,8 @@ transport: ignore_type_cast_errors: "${SNMP_RESPONSE_IGNORE_TYPE_CAST_ERRORS:false}" # Thread pool size for scheduler that executes device querying tasks scheduler_thread_pool_size: "${SNMP_SCHEDULER_THREAD_POOL_SIZE:4}" + # Maximum number of retry attempts for SNMP bootstrap during startup + bootstrap_retries: "${SNMP_BOOTSTRAP_RETRIES:8}" stats: # Enable/Disable the collection of transport statistics enabled: "${TB_TRANSPORT_STATS_ENABLED:true}" diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index 56b4e00162..0cbdc3bd50 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -18,6 +18,7 @@ package org.thingsboard.server.transport.snmp; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.event.EventListener; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Device; @@ -53,9 +54,9 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedDeque; @TbSnmpTransportComponent @Component @@ -72,14 +73,35 @@ public class SnmpTransportContext extends TransportContext { private final SnmpAuthService snmpAuthService; private final Map sessions = new ConcurrentHashMap<>(); - private final Collection allSnmpDevicesIds = new ConcurrentLinkedDeque<>(); + private final Set allSnmpDevicesIds = ConcurrentHashMap.newKeySet(); + + @Value("${transport.snmp.bootstrap_retries}") + private int snmpBootstrapMaxRetries; @AfterStartUp(order = AfterStartUp.AFTER_TRANSPORT_SERVICE) public void fetchDevicesAndEstablishSessions() { - log.info("Initializing SNMP devices sessions"); + getExecutor().execute(this::bootstrapWithRetries); + } + private void bootstrapWithRetries() { + for (int attempt = 1; attempt <= snmpBootstrapMaxRetries; attempt++) { + try { + doBootstrap(); + return; + } catch (Exception e) { + if (attempt >= snmpBootstrapMaxRetries) { + log.error("SNMP bootstrap failed after {} attempts.", attempt, e); + return; + } + log.warn("SNMP bootstrap attempt {}/{} failed. Retrying immediately...", attempt, snmpBootstrapMaxRetries, e); + } + } + } + + private void doBootstrap() { + log.info("Initializing SNMP devices sessions"); int batchIndex = 0; - int batchSize = 512; + final int batchSize = 512; boolean nextBatchExists = true; while (nextBatchExists) { @@ -89,13 +111,17 @@ public class SnmpTransportContext extends TransportContext { .peek(allSnmpDevicesIds::add) .filter(deviceId -> balancingService.isManagedByCurrentTransport(deviceId.getId())) .map(protoEntityService::getDeviceById) - .forEach(device -> getExecutor().execute(() -> establishDeviceSession(device))); + .forEach(device -> { + if (!sessions.containsKey(device.getId())) { + getExecutor().execute(() -> establishDeviceSession(device)); + } + }); nextBatchExists = snmpDevicesResponse.getHasNextPage(); batchIndex++; } - log.debug("Found all SNMP devices ids: {}", allSnmpDevicesIds); + log.debug("Found all SNMP devices ids: {}", allSnmpDevicesIds.size()); } private void establishDeviceSession(Device device) { From 683879b2ce6628036c14d98491058934fb3b8fa0 Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Wed, 12 Nov 2025 23:42:57 +0200 Subject: [PATCH 2/5] fixed --- .../server/transport/snmp/SnmpTransportContext.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index 0cbdc3bd50..ed8bfb932a 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -21,6 +21,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.event.EventListener; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceTransportType; @@ -57,6 +58,8 @@ import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; @TbSnmpTransportComponent @Component @@ -74,13 +77,14 @@ public class SnmpTransportContext extends TransportContext { private final Map sessions = new ConcurrentHashMap<>(); private final Set allSnmpDevicesIds = ConcurrentHashMap.newKeySet(); + private final ExecutorService snmpExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("snmp-bootstrap")); @Value("${transport.snmp.bootstrap_retries}") private int snmpBootstrapMaxRetries; @AfterStartUp(order = AfterStartUp.AFTER_TRANSPORT_SERVICE) public void fetchDevicesAndEstablishSessions() { - getExecutor().execute(this::bootstrapWithRetries); + snmpExecutor.execute(this::bootstrapWithRetries); } private void bootstrapWithRetries() { From f6136acc523dfaa6201e809dc177d27f03a817ad Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Wed, 12 Nov 2025 23:50:09 +0200 Subject: [PATCH 3/5] refactoring --- .../server/transport/snmp/SnmpTransportContext.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index ed8bfb932a..3912e79fcc 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -105,7 +105,7 @@ public class SnmpTransportContext extends TransportContext { private void doBootstrap() { log.info("Initializing SNMP devices sessions"); int batchIndex = 0; - final int batchSize = 512; + int batchSize = 512; boolean nextBatchExists = true; while (nextBatchExists) { @@ -125,7 +125,7 @@ public class SnmpTransportContext extends TransportContext { batchIndex++; } - log.debug("Found all SNMP devices ids: {}", allSnmpDevicesIds.size()); + log.debug("Found all SNMP devices ids: {}", allSnmpDevicesIds); } private void establishDeviceSession(Device device) { From 7dbea4a756e217f0a972a54bce40025a958b57c9 Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Thu, 13 Nov 2025 16:26:09 +0200 Subject: [PATCH 4/5] refactoring --- .../src/main/resources/thingsboard.yml | 4 +- .../transport/snmp/SnmpTransportContext.java | 72 ++++++++++--------- 2 files changed, 42 insertions(+), 34 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 81fe17c2b6..dfdd3a407c 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1295,8 +1295,8 @@ transport: ignore_type_cast_errors: "${SNMP_RESPONSE_IGNORE_TYPE_CAST_ERRORS:false}" # Thread pool size for scheduler that executes device querying tasks scheduler_thread_pool_size: "${SNMP_SCHEDULER_THREAD_POOL_SIZE:4}" - # Maximum number of retry attempts for SNMP bootstrap during startup - bootstrap_retries: "${SNMP_BOOTSTRAP_RETRIES:8}" + # Maximum number of retry attempts for a single SNMP devices batch during bootstrap. + batch_retries: "${SNMP_BOOTSTRAP_RETRIES:8}" stats: # Enable/Disable the collection of transport statistics enabled: "${TB_TRANSPORT_STATS_ENABLED:true}" diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index 3912e79fcc..02140a982b 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.transport.snmp; +import jakarta.annotation.PreDestroy; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -60,6 +61,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; @TbSnmpTransportComponent @Component @@ -79,8 +81,8 @@ public class SnmpTransportContext extends TransportContext { private final Set allSnmpDevicesIds = ConcurrentHashMap.newKeySet(); private final ExecutorService snmpExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("snmp-bootstrap")); - @Value("${transport.snmp.bootstrap_retries}") - private int snmpBootstrapMaxRetries; + @Value("${transport.snmp.batch_retries}") + private int snmpBootstrapBatchRetries; @AfterStartUp(order = AfterStartUp.AFTER_TRANSPORT_SERVICE) public void fetchDevicesAndEstablishSessions() { @@ -88,44 +90,45 @@ public class SnmpTransportContext extends TransportContext { } private void bootstrapWithRetries() { - for (int attempt = 1; attempt <= snmpBootstrapMaxRetries; attempt++) { - try { - doBootstrap(); - return; - } catch (Exception e) { - if (attempt >= snmpBootstrapMaxRetries) { - log.error("SNMP bootstrap failed after {} attempts.", attempt, e); - return; - } - log.warn("SNMP bootstrap attempt {}/{} failed. Retrying immediately...", attempt, snmpBootstrapMaxRetries, e); - } - } - } - - private void doBootstrap() { log.info("Initializing SNMP devices sessions"); int batchIndex = 0; int batchSize = 512; boolean nextBatchExists = true; while (nextBatchExists) { - TransportProtos.GetSnmpDevicesResponseMsg snmpDevicesResponse = protoEntityService.getSnmpDevicesIds(batchIndex, batchSize); - snmpDevicesResponse.getIdsList().stream() - .map(id -> new DeviceId(UUID.fromString(id))) - .peek(allSnmpDevicesIds::add) - .filter(deviceId -> balancingService.isManagedByCurrentTransport(deviceId.getId())) - .map(protoEntityService::getDeviceById) - .forEach(device -> { - if (!sessions.containsKey(device.getId())) { - getExecutor().execute(() -> establishDeviceSession(device)); - } - }); + for (int attempt = 1; attempt <= snmpBootstrapBatchRetries; attempt++) { + try { + TransportProtos.GetSnmpDevicesResponseMsg snmpDevicesResponse = protoEntityService.getSnmpDevicesIds(batchIndex, batchSize); + snmpDevicesResponse.getIdsList().stream() + .map(id -> new DeviceId(UUID.fromString(id))) + .peek(allSnmpDevicesIds::add) + .filter(deviceId -> balancingService.isManagedByCurrentTransport(deviceId.getId())) + .map(protoEntityService::getDeviceById) + .forEach(device -> getExecutor().execute(() -> establishDeviceSession(device))); + nextBatchExists = snmpDevicesResponse.getHasNextPage(); + batchIndex++; + break; + } catch (Exception e) { + if (e instanceof InterruptedException) { + log.warn("SNMP bootstrap interrupted. Stopping bootstrap task.", e); + return; + } - nextBatchExists = snmpDevicesResponse.getHasNextPage(); - batchIndex++; + if (attempt >= snmpBootstrapBatchRetries) { + log.error("SNMP bootstrap: batch {} failed after {} attempts.", batchIndex, attempt, e); + return; + } + log.warn("SNMP bootstrap: batch {} attempt {}/{} failed.", batchIndex, attempt, snmpBootstrapBatchRetries, e); + try { + TimeUnit.SECONDS.sleep(10); + } catch (InterruptedException ex) { + log.warn("SNMP bootstrap interrupted. Stopping bootstrap task."); + return; + } + } + } } - - log.debug("Found all SNMP devices ids: {}", allSnmpDevicesIds); + log.debug("Found SNMP devices ids: {}", allSnmpDevicesIds); } private void establishDeviceSession(Device device) { @@ -330,4 +333,9 @@ public class SnmpTransportContext extends TransportContext { return sessions.values(); } + @PreDestroy + public void destroy() { + snmpExecutor.shutdown(); + } + } From 25181bb1c88ac275fed13d8020f2a97380a2b36b Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Fri, 14 Nov 2025 11:21:35 +0200 Subject: [PATCH 5/5] refactoring --- .../server/transport/snmp/SnmpTransportContext.java | 5 ----- transport/snmp/src/main/resources/tb-snmp-transport.yml | 2 ++ 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index 02140a982b..93ce3d1278 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -109,11 +109,6 @@ public class SnmpTransportContext extends TransportContext { batchIndex++; break; } catch (Exception e) { - if (e instanceof InterruptedException) { - log.warn("SNMP bootstrap interrupted. Stopping bootstrap task.", e); - return; - } - if (attempt >= snmpBootstrapBatchRetries) { log.error("SNMP bootstrap: batch {} failed after {} attempts.", batchIndex, attempt, e); return; diff --git a/transport/snmp/src/main/resources/tb-snmp-transport.yml b/transport/snmp/src/main/resources/tb-snmp-transport.yml index 79aee31921..567654cce4 100644 --- a/transport/snmp/src/main/resources/tb-snmp-transport.yml +++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml @@ -151,6 +151,8 @@ transport: ignore_type_cast_errors: "${SNMP_RESPONSE_IGNORE_TYPE_CAST_ERRORS:false}" # Thread pool size for scheduler that executes device querying tasks scheduler_thread_pool_size: "${SNMP_SCHEDULER_THREAD_POOL_SIZE:4}" + # Maximum number of retry attempts for a single SNMP devices batch during bootstrap. + batch_retries: "${SNMP_BOOTSTRAP_RETRIES:8}" sessions: # Session inactivity timeout is a global configuration parameter that defines how long the device transport session will be opened after the last message arrives from the device. # The parameter value is in milliseconds.