Browse Source
Replace the blocking semaphore guard with a non-blocking bounded FIFO queue + semaphore pattern: - No semaphore/queue when maxParallelRequestsCount=0 (default): direct doHttpCall, identical to the old behavior. - When a concurrency limit is set, incoming messages are enqueued via non-blocking offer(); a full queue triggers onFailure immediately. - tryProcess() acquires one semaphore slot and dispatches the next valid queued task. Stale tasks (batch deadline expired) are dropped and the slot reused in the same pass. - doFinally hook releases the semaphore and calls tryProcess() exactly once after any terminal signal (success, error, cancel), preventing double-release and permit leaks. - publishOn(externalCallExecutor) moves callbacks off reactor-netty I/O threads. System-level safety caps are wired through thingsboard.yml → ActorSystemContext → TbContext → TbHttpClient, scoped to rule-engine services only via @TbRuleEngineComponent: actors.rule.external.http_client.max_parallel_requests (ACTORS_RULE_EXTERNAL_HTTP_CLIENT_MAX_PARALLEL_REQUESTS) actors.rule.external.http_client.max_pending_requests (ACTORS_RULE_EXTERNAL_HTTP_CLIENT_MAX_PENDING_REQUESTS) actors.rule.external.http_client.pool_max_connections (ACTORS_RULE_EXTERNAL_HTTP_CLIENT_POOL_MAX_CONNECTIONS) Backward compat: TB_RE_HTTP_CLIENT_POOL_MAX_CONNECTIONS still honored via yaml fallback. Observability: five AtomicLong counters (dispatched, success, failure, droppedQueueFull, droppedStale) with periodic WARN anomaly logging including semaphorePermits for leak detection. No configuration changes or upgrade scripts required — docker image update is sufficient. Rename RestApiCallNodeSettings to TbHttpClientSettings The settings are about HTTP client transport concerns (connection pool, concurrency, queue depth), not REST API Call node business logic. The new name matches the consumer (TbHttpClient) and the YAML path (actors.rule.external.http_client.*).pull/15334/head
Sergey Matvienko
2 months ago
10 changed files with 454 additions and 35 deletions
@ -0,0 +1,51 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed 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.thingsboard.server.config; |
|||
|
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.rule.engine.api.TbHttpClientSettings; |
|||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|||
|
|||
@TbRuleEngineComponent |
|||
@Component |
|||
public class TbHttpClientSettingsComponent implements TbHttpClientSettings { |
|||
|
|||
@Value("${actors.rule.external.http_client.max_parallel_requests:0}") |
|||
private int maxParallelRequests; |
|||
|
|||
@Value("${actors.rule.external.http_client.max_pending_requests:0}") |
|||
private int maxPendingRequests; |
|||
|
|||
@Value("${actors.rule.external.http_client.pool_max_connections:0}") |
|||
private int poolMaxConnections; |
|||
|
|||
@Override |
|||
public int getMaxParallelRequests() { |
|||
return maxParallelRequests; |
|||
} |
|||
|
|||
@Override |
|||
public int getMaxPendingRequests() { |
|||
return maxPendingRequests; |
|||
} |
|||
|
|||
@Override |
|||
public int getPoolMaxConnections() { |
|||
return poolMaxConnections; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,49 @@ |
|||
/** |
|||
* Copyright © 2016-2026 The Thingsboard Authors |
|||
* |
|||
* Licensed 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.thingsboard.rule.engine.api; |
|||
|
|||
/** |
|||
* Server-level safety caps for the HTTP client used by the REST API Call rule node. |
|||
* Values are read from {@code thingsboard.yml} (or the corresponding environment variables) |
|||
* and applied as hard ceilings on top of the per-node tenant configuration. |
|||
* A value of {@code 0} means no system-level restriction. |
|||
*/ |
|||
public interface TbHttpClientSettings { |
|||
|
|||
/** System ceiling for {@code maxParallelRequestsCount}. 0 = no system limit. */ |
|||
int getMaxParallelRequests(); |
|||
|
|||
/** System ceiling for the pending-request queue depth. 0 = no system limit. */ |
|||
int getMaxPendingRequests(); |
|||
|
|||
/** |
|||
* Maximum number of TCP connections in the reactor-netty pool per node instance. |
|||
* 0 = use reactor-netty's default: {@code max(availableProcessors, 8) * 2}. |
|||
*/ |
|||
int getPoolMaxConnections(); |
|||
|
|||
TbHttpClientSettings DEFAULT = new TbHttpClientSettings() { |
|||
@Override |
|||
public int getMaxParallelRequests() { return 0; } |
|||
|
|||
@Override |
|||
public int getMaxPendingRequests() { return 0; } |
|||
|
|||
@Override |
|||
public int getPoolMaxConnections() { return 0; } |
|||
}; |
|||
|
|||
} |
|||
Loading…
Reference in new issue