From a86eb53038e3c5e74bb7bc16d81cbbfddf49a8a4 Mon Sep 17 00:00:00 2001 From: Viacheslav Klimov Date: Fri, 1 Jul 2022 15:31:29 +0300 Subject: [PATCH] Keep data order when getting chunked messages --- .../sync/vc/DefaultEntitiesVersionControlService.java | 6 +++--- .../sync/vc/DefaultGitVersionControlQueueService.java | 7 ++++--- .../server/common/data/sync/vc/EntityLoadError.java | 11 ++++++++--- 3 files changed, 15 insertions(+), 9 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultEntitiesVersionControlService.java b/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultEntitiesVersionControlService.java index 00e98cd3a1..005e90001b 100644 --- a/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultEntitiesVersionControlService.java +++ b/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultEntitiesVersionControlService.java @@ -289,7 +289,7 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont return cachePut(ctx.getRequestId(), onError(e.getExternalId(), e.getCause())); } catch (Exception e) { log.info("[{}] Failed to process request [{}] due to: ", ctx.getTenantId(), request, e); - return cachePut(ctx.getRequestId(), VersionLoadResult.error(EntityLoadError.runtimeError(e.getMessage()))); + return cachePut(ctx.getRequestId(), VersionLoadResult.error(EntityLoadError.runtimeError(e))); } } @@ -433,7 +433,7 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont } private VersionLoadResult onError(EntityId externalId, Throwable e) { - return analyze(e, externalId).orElse(VersionLoadResult.error(EntityLoadError.runtimeError(e.getMessage()))); + return analyze(e, externalId).orElse(VersionLoadResult.error(EntityLoadError.runtimeError(e))); } private Optional analyze(Throwable e, EntityId externalId) { @@ -605,7 +605,7 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont private void processLoadError(EntitiesImportCtx ctx, Throwable e) { log.debug("[{}] Failed to load the commit: {}", ctx.getRequestId(), ctx.getVersionId(), e); - cachePut(ctx.getRequestId(), VersionLoadResult.error(EntityLoadError.runtimeError(e.getMessage()))); + cachePut(ctx.getRequestId(), VersionLoadResult.error(EntityLoadError.runtimeError(e))); } private void cachePut(UUID requestId, VersionCreationResult result) { diff --git a/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultGitVersionControlQueueService.java b/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultGitVersionControlQueueService.java index c9b620ad1c..0c098d8281 100644 --- a/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultGitVersionControlQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultGitVersionControlQueueService.java @@ -77,6 +77,7 @@ import org.thingsboard.server.service.sync.vc.data.VoidGitRequest; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -101,7 +102,7 @@ public class DefaultGitVersionControlQueueService implements GitVersionControlQu private final SchedulerComponent scheduler; private final Map> pendingRequestMap = new HashMap<>(); - private final Map> chunkedMsgs = new ConcurrentHashMap<>(); + private final Map> chunkedMsgs = new ConcurrentHashMap<>(); @Value("${queue.vc.request-timeout:60000}") private int requestTimeout; @@ -285,7 +286,7 @@ public class DefaultGitVersionControlQueueService implements GitVersionControlQu @SuppressWarnings("rawtypes") public ListenableFuture getEntity(TenantId tenantId, String versionId, EntityId entityId) { EntityContentGitRequest request = new EntityContentGitRequest(tenantId, versionId, entityId); - chunkedMsgs.put(request.getRequestId(), new HashMap<>()); + chunkedMsgs.put(request.getRequestId(), new LinkedHashMap<>()); registerAndSend(request, builder -> builder.setEntityContentRequest(EntityContentRequestMsg.newBuilder() .setVersionId(versionId) .setEntityType(entityId.getEntityType().name()) @@ -327,7 +328,7 @@ public class DefaultGitVersionControlQueueService implements GitVersionControlQu @SuppressWarnings("rawtypes") public ListenableFuture> getEntities(TenantId tenantId, String versionId, EntityType entityType, int offset, int limit) { EntitiesContentGitRequest request = new EntitiesContentGitRequest(tenantId, versionId, entityType); - chunkedMsgs.put(request.getRequestId(), new HashMap<>()); + chunkedMsgs.put(request.getRequestId(), new LinkedHashMap<>()); registerAndSend(request, builder -> builder.setEntitiesContentRequest(EntitiesContentRequestMsg.newBuilder() .setVersionId(versionId) .setEntityType(entityType.name()) diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/sync/vc/EntityLoadError.java b/common/data/src/main/java/org/thingsboard/server/common/data/sync/vc/EntityLoadError.java index cd3dc5e0e2..cbfe9f6457 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/sync/vc/EntityLoadError.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/sync/vc/EntityLoadError.java @@ -18,10 +18,11 @@ package org.thingsboard.server.common.data.sync.vc; import com.fasterxml.jackson.annotation.JsonInclude; import lombok.Builder; import lombok.Data; +import org.apache.commons.lang3.ClassUtils; +import org.apache.commons.lang3.StringUtils; import org.thingsboard.server.common.data.id.EntityId; import java.io.Serializable; -import java.util.List; @Data @Builder @@ -43,8 +44,12 @@ public class EntityLoadError implements Serializable { return EntityLoadError.builder().type("MISSING_REFERENCED_ENTITY").source(sourceId).target(targetId).build(); } - public static EntityLoadError runtimeError(String msg) { - return EntityLoadError.builder().type("RUNTIME").message(msg).build(); + public static EntityLoadError runtimeError(Throwable e) { + String message = e.getMessage(); + if (StringUtils.isEmpty(message)) { + message = "unexpected error (" + ClassUtils.getShortClassName(e.getClass()) + ")"; + } + return EntityLoadError.builder().type("RUNTIME").message(message).build(); } }