refactor

littlehorse-enterprises · Feb 19, 2025 · 1a7d2ec · 1a7d2ec
1 parent 1aecebe
commit 1a7d2ec
Showing 1 changed file with 12 additions and 9 deletions.
diff --git a/...rc/main/java/io/littlehorse/server/streams/topology/core/processors/CommandProcessor.java b/...rc/main/java/io/littlehorse/server/streams/topology/core/processors/CommandProcessor.java
@@ -175,7 +175,7 @@ private void forwardMetricsUpdates(long timestamp) {
         PartitionMetricInventoryModel metricInventory = clusterStore.get(
                 PartitionMetricInventoryModel.METRIC_INVENTORY_STORE_KEY, PartitionMetricInventoryModel.class);
         if (metricInventory != null) {
-            Map<TenantIdModel, AggregateMetricsModel> commandsPerTenant = new HashMap<>();
+            Map<TenantIdModel, List<AggregateMetricsModel>> commandsPerTenant = new HashMap<>();
             for (PartitionMetricIdModel partitionMetricId : metricInventory.getMetrics()) {
                 TenantScopedStore tenantStore = TenantScopedStore.newInstance(
                         nativeStore, partitionMetricId.getTenantId(), new BackgroundContext());
@@ -187,17 +187,20 @@ private void forwardMetricsUpdates(long timestamp) {
                 List<RepartitionWindowedMetricModel> windowedMetrics =
                         partitionMetric.buildRepartitionCommand(LocalDateTime.now());
                 tenantStore.put(new StoredGetable<>(partitionMetric));
-                AggregateMetricsModel current = commandsPerTenant.getOrDefault(
+                List<AggregateMetricsModel> metricsPerTenant =
+                        commandsPerTenant.getOrDefault(partitionMetricId.getTenantId(), new ArrayList<>());
+                AggregateMetricsModel current = new AggregateMetricsModel(
                         partitionMetricId.getTenantId(),
-                        new AggregateMetricsModel(
-                                partitionMetricId.getTenantId(),
-                                partitionMetricId.getMetricId(),
-                                new ArrayList<>(),
-                                ctx.taskId().partition()));
+                        partitionMetricId.getMetricId(),
+                        new ArrayList<>(),
+                        ctx.taskId().partition());
                 current.addWindowedMetric(windowedMetrics);
-                commandsPerTenant.putIfAbsent(partitionMetricId.getTenantId(), current);
+                metricsPerTenant.add(current);
+                commandsPerTenant.putIfAbsent(partitionMetricId.getTenantId(), metricsPerTenant);
             }
-            forwardRepartitionCommands(commandsPerTenant.values());
+            forwardRepartitionCommands(commandsPerTenant.values().stream()
+                    .flatMap(Collection::stream)
+                    .toList());
         }
     }