Skip to content

Commit e5357d1

Browse files
volodymyr-babakvolodymyr-babak
committed
Merge remote-tracking branch 'origin/rc' into release-4.2
2 parents 8491ec5 + d2e69c3 commit e5357d1

File tree

5 files changed

+94
-41
lines changed

5 files changed

+94
-41
lines changed

‎application/src/main/java/org/thingsboard/server/service/cloud/BaseCloudManagerService.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -712,7 +712,7 @@ protected ListenableFuture<Boolean> processCloudEvents(List<CloudEvent> cloudEve
712712
.toList();
713713

714714
if (uplinkMsgPack.isEmpty()) {
715-
return Futures.immediateFuture(true);
715+
return Futures.immediateFuture(false);
716716
}
717717

718718
processMsgPack(uplinkMsgPack, isGeneralMsg);

‎application/src/main/java/org/thingsboard/server/service/cloud/KafkaCloudManagerService.java‎

Lines changed: 49 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@
3434
import org.thingsboard.server.queue.settings.TbQueueCloudEventSettings;
3535
import org.thingsboard.server.queue.settings.TbQueueCloudEventTSSettings;
3636

37-
import java.util.ArrayList;
3837
import java.util.List;
3938
import java.util.concurrent.ExecutorService;
4039
import java.util.concurrent.Executors;
@@ -118,48 +117,74 @@ protected void onDestroy() throws InterruptedException {
118117
}
119118
}
120119

121-
private void processUplinkMessages(List<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> consumer) {
122-
log.trace("[{}] starting processing cloud events", tenantId);
123-
if (initialized && !syncInProgress) {
124-
isGeneralProcessInProgress = true;
125-
processMessages(msgs, consumer, true);
126-
isGeneralProcessInProgress = false;
127-
} else {
128-
sleep();
129-
}
120+
private void processUplinkMessages(List<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> msgs,
121+
TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> consumer) {
122+
boolean isProcessed = false;
123+
do {
124+
log.trace("[{}] Trying to process general uplink messages", tenantId);
125+
126+
if (initialized && !syncInProgress) {
127+
isGeneralProcessInProgress = true;
128+
isProcessed = processMessages(msgs, consumer, true);
129+
isGeneralProcessInProgress = false;
130+
} else {
131+
log.debug("[{}] Waiting: initialized={}, syncInProgress={}", tenantId, initialized, syncInProgress);
132+
}
133+
134+
if (!isProcessed) {
135+
sleep();
136+
}
137+
} while (!isProcessed);
130138
}
131139

132-
private void processTsUplinkMessages(List<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> consumer) {
133-
if (initialized && !syncInProgress && !isGeneralProcessInProgress) {
134-
processMessages(msgs, consumer, false);
135-
} else {
136-
sleep();
137-
}
140+
private void processTsUplinkMessages(List<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> msgs,
141+
TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> consumer) {
142+
boolean isProcessed = false;
143+
144+
do {
145+
log.trace("[{}] Trying to process TS uplink messages", tenantId);
146+
147+
if (initialized && !syncInProgress && !isGeneralProcessInProgress) {
148+
isProcessed = processMessages(msgs, consumer, false);
149+
} else {
150+
log.debug("[{}] Waiting: initialized={}, syncInProgress={}, generalInProgress={}",
151+
tenantId, initialized, syncInProgress, isGeneralProcessInProgress);
152+
}
153+
154+
if (!isProcessed) {
155+
sleep();
156+
}
157+
} while (!isProcessed);
138158
}
139159

140160
private void sleep() {
141161
try {
142162
Thread.sleep(cloudEventStorageSettings.getNoRecordsSleepInterval());
143163
} catch (InterruptedException interruptedException) {
144-
log.trace("Failed to wait until the server has capacity to handle new requests", interruptedException);
164+
log.trace("Interrupted while waiting to retry uplink processing", interruptedException);
145165
}
146166
}
147167

148-
private void processMessages(List<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> consumer, boolean isGeneralMsg) {
149-
List<CloudEvent> cloudEvents = new ArrayList<>();
150-
for (TbProtoQueueMsg<TransportProtos.ToCloudEventMsg> msg : msgs) {
151-
CloudEvent cloudEvent = ProtoUtils.fromProto(msg.getValue().getCloudEventMsg());
152-
cloudEvents.add(cloudEvent);
153-
}
168+
private boolean processMessages(List<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> msgs,
169+
TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCloudEventMsg>> consumer,
170+
boolean isGeneralMsg) {
171+
List<CloudEvent> cloudEvents = msgs.stream()
172+
.map(msg -> ProtoUtils.fromProto(msg.getValue().getCloudEventMsg()))
173+
.toList();
174+
154175
try {
155176
boolean isInterrupted = processCloudEvents(cloudEvents, isGeneralMsg).get();
156177
if (isInterrupted) {
157-
log.debug("[{}] Send uplink messages task was interrupted", tenantId);
178+
log.warn("[{}] Send uplink messages task was interrupted", tenantId);
179+
return false;
158180
} else {
159181
consumer.commit();
182+
log.trace("[{}] Successfully processed {} uplink messages (type={})", tenantId, cloudEvents.size(), isGeneralMsg ? "GENERAL" : "TS");
183+
return true;
160184
}
161185
} catch (Exception e) {
162186
log.error("Failed to process all uplink messages", e);
187+
return false;
163188
}
164189
}
165190

‎application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java‎

Lines changed: 40 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
import org.thingsboard.server.queue.provider.TbCoreQueueFactory;
3737
import org.thingsboard.server.service.edge.EdgeContextComponent;
3838

39-
import java.util.ArrayList;
4039
import java.util.List;
4140
import java.util.UUID;
4241
import java.util.concurrent.ExecutorService;
@@ -46,7 +45,6 @@
4645

4746
@Slf4j
4847
public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
49-
5048
private final TopicService topicService;
5149
private final TbCoreQueueFactory tbCoreQueueFactory;
5250
private final KafkaAdmin kafkaAdmin;
@@ -68,29 +66,55 @@ public KafkaEdgeGrpcSession(EdgeContextComponent ctx, TopicService topicService,
6866
this.kafkaAdmin = kafkaAdmin;
6967
}
7068

71-
private void processMsgs(List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> consumer) {
72-
log.trace("[{}][{}] starting processing edge events", tenantId, edge.getId());
73-
if (!isConnected() || isSyncInProgress() || isHighPriorityProcessing) {
74-
log.debug("[{}][{}] edge not connected, edge sync is not completed or high priority processing in progress, " +
75-
"connected = {}, sync in progress = {}, high priority in progress = {}. Skipping iteration",
76-
tenantId, edge.getId(), isConnected(), isSyncInProgress(), isHighPriorityProcessing);
77-
return;
78-
}
79-
List<EdgeEvent> edgeEvents = new ArrayList<>();
80-
for (TbProtoQueueMsg<ToEdgeEventNotificationMsg> msg : msgs) {
81-
EdgeEvent edgeEvent = ProtoUtils.fromProto(msg.getValue().getEdgeEventMsg());
82-
edgeEvents.add(edgeEvent);
83-
}
69+
private void processMsgs(List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> consumer) throws InterruptedException {
70+
boolean isProcessed = false;
71+
72+
do {
73+
log.trace("[{}][{}] Trying to process edge events", tenantId, edge.getId());
74+
75+
if (isConnected() && !isSyncInProgress() && !isHighPriorityProcessing) {
76+
isProcessed = tryProcessMsgs(msgs, consumer);
77+
} else {
78+
log.debug("[{}][{}] edge not connected, edge sync is not completed or high priority processing in progress, " +
79+
"connected = {}, sync in progress = {}, high priority in progress = {}. Skipping iteration",
80+
tenantId, edge.getId(), isConnected(), isSyncInProgress(), isHighPriorityProcessing);
81+
}
82+
83+
if (!isProcessed) {
84+
sleep();
85+
}
86+
} while (!isProcessed);
87+
}
88+
89+
private boolean tryProcessMsgs(List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> msgs,
90+
TbQueueConsumer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> consumer) {
91+
List<EdgeEvent> edgeEvents = msgs.stream()
92+
.map(msg -> ProtoUtils.fromProto(msg.getValue().getEdgeEventMsg()))
93+
.toList();
94+
8495
List<DownlinkMsg> downlinkMsgsPack = convertToDownlinkMsgsPack(edgeEvents);
96+
8597
try {
8698
boolean isInterrupted = sendDownlinkMsgsPack(downlinkMsgsPack).get();
8799
if (isInterrupted) {
88-
log.debug("[{}][{}] Send downlink messages task was interrupted", tenantId, edge.getId());
100+
log.warn("[{}][{}] Send downlink messages task was interrupted", tenantId, edge.getId());
101+
return false;
89102
} else {
90103
consumer.commit();
104+
log.trace("[{}][{}] Successfully processed {} edge events", tenantId, edge.getId(), edgeEvents.size());
105+
return true;
91106
}
92107
} catch (Exception e) {
93108
log.error("[{}][{}] Failed to process downlink messages", tenantId, edge.getId(), e);
109+
return false;
110+
}
111+
}
112+
113+
private void sleep() {
114+
try {
115+
Thread.sleep(ctx.getEdgeEventStorageSettings().getNoRecordsSleepInterval());
116+
} catch (InterruptedException interruptedException) {
117+
log.trace("[{}][{}] Interrupted while waiting to retry edge events processing", tenantId, edge.getId(), interruptedException);
94118
}
95119
}
96120

‎common/data/src/main/java/org/thingsboard/server/common/data/cloud/CloudEventType.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ public enum CloudEventType {
2929
ALARM(EntityType.ALARM),
3030
ALARM_COMMENT(null),
3131
RULE_CHAIN(EntityType.RULE_CHAIN),
32+
RULE_NODE(EntityType.RULE_NODE),
3233
RULE_CHAIN_METADATA(null),
3334
USER(EntityType.USER),
3435
TENANT(EntityType.TENANT),

‎ui-ngx/src/app/core/services/menu.models.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -774,6 +774,9 @@ const menuFilters = new Map<MenuId, MenuFilter>([
774774
],
775775
[
776776
MenuId.mobile_bundles, () => false
777+
],
778+
[
779+
MenuId.ai_models, () => false
777780
]
778781
]);
779782

0 commit comments

Comments
 (0)