Simplify the batchMetrics function

Author: psx95

CHANGELOG.md

@@ -11,6 +11,8 @@
 
 #### Incubating
 
+* Add support for configuring `setMaxExportBatchSize` in `PeriodicMetricReader` ([#8296](https://github.com/open-telemetry/opentelemetry-java/pull/8296))
+
 * **BREAKING** Update `EnvironmentGetter` and `EnvironmentSetter` key normalization to reflect spec
   changes
   ([#8233](https://github.com/open-telemetry/opentelemetry-java/pull/8233))

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/MetricExportBatcher.java

@@ -66,66 +66,35 @@ Collection<Collection<MetricData>> batchMetrics(Collection<MetricData> metrics)
       return Collections.emptyList();
     }
     Collection<Collection<MetricData>> preparedBatchesForExport = new ArrayList<>();
-    BatchState currentBatch = new BatchState(new ArrayList<>(maxExportBatchSize), 0);
-
-    // Fill active batch and split overlapping metric points if needed
+    List<MetricData> currentBatch = new ArrayList<>(maxExportBatchSize);
+    int currentPointsInBatch = 0;
     for (MetricData metricData : metrics) {
-      MetricDataSplitOperationResult splitResult = prepareExportBatches(metricData, currentBatch);
-      preparedBatchesForExport.addAll(splitResult.getPreparedBatches());
-      currentBatch = splitResult.getLastInProgressBatch();
-    }
-
-    // Push trailing capacity block
-    if (!currentBatch.metrics.isEmpty()) {
-      preparedBatchesForExport.add(currentBatch.metrics);
-    }
-    return Collections.unmodifiableCollection(preparedBatchesForExport);
-  }
-
-  /**
-   * Prepares export batches from a single metric data object. This function only operates on a
-   * single metric data object, fills up the current batch with as many points as possible from the
-   * metric data object, and then creates new metric data objects for the remaining points.
-   *
-   * @param metricData The metric data object to split.
-   * @param currentBatch The current batch of metric data objects.
-   * @return A result containing the prepared batches and the last in-progress batch.
-   */
-  private MetricDataSplitOperationResult prepareExportBatches(
-      MetricData metricData, BatchState currentBatch) {
-    int remainingCapacityInCurrentBatch = maxExportBatchSize - currentBatch.points;
-    int totalPointsInMetricData = metricData.getData().getPoints().size();
-
-    if (remainingCapacityInCurrentBatch >= totalPointsInMetricData) {
-      currentBatch.metrics.add(metricData);
-      currentBatch.points += totalPointsInMetricData;
-      return new MetricDataSplitOperationResult(Collections.emptyList(), currentBatch);
-    } else {
-      // Remaining capacity can't hold all points, partition existing metric data object
-      List<PointData> originalPointsList = new ArrayList<>(metricData.getData().getPoints());
-      Collection<Collection<MetricData>> preparedBatches = new ArrayList<>();
+      int totalPointsInMetric = metricData.getData().getPoints().size();
+      if (currentPointsInBatch + totalPointsInMetric <= maxExportBatchSize) {
+        currentBatch.add(metricData);
+        currentPointsInBatch += totalPointsInMetric;
+        continue;
+      }
       int currentIndex = 0;
-
-      while (currentIndex < totalPointsInMetricData) {
-        int pointsToTake =
-            Math.min(totalPointsInMetricData - currentIndex, remainingCapacityInCurrentBatch);
-
-        if (pointsToTake > 0) {
-          currentBatch.metrics.add(
-              copyMetricData(metricData, originalPointsList, currentIndex, pointsToTake));
-          currentBatch.points += pointsToTake;
-          currentIndex += pointsToTake;
-          remainingCapacityInCurrentBatch -= pointsToTake;
-        }
-
-        if (remainingCapacityInCurrentBatch == 0) {
-          preparedBatches.add(currentBatch.metrics);
-          currentBatch = new BatchState(new ArrayList<>(maxExportBatchSize), 0);
-          remainingCapacityInCurrentBatch = maxExportBatchSize;
+      List<PointData> originalPointsList = new ArrayList<>(metricData.getData().getPoints());
+      while (currentIndex < totalPointsInMetric) {
+        if (currentPointsInBatch == maxExportBatchSize) {
+          preparedBatchesForExport.add(currentBatch);
+          currentBatch = new ArrayList<>(maxExportBatchSize);
+          currentPointsInBatch = 0;
         }
+        int pointsToTake =
+            Math.min(maxExportBatchSize - currentPointsInBatch, totalPointsInMetric - currentIndex);
+        currentBatch.add(
+            copyMetricData(metricData, originalPointsList, currentIndex, pointsToTake));
+        currentPointsInBatch += pointsToTake;
+        currentIndex += pointsToTake;
       }
-      return new MetricDataSplitOperationResult(preparedBatches, currentBatch);
     }
+    if (!currentBatch.isEmpty()) {
+      preparedBatchesForExport.add(currentBatch);
+    }
+    return Collections.unmodifiableCollection(preparedBatchesForExport);
   }
 
   private static MetricData copyMetricData(
@@ -224,57 +193,4 @@ private static MetricData createMetricDataWithPoints(
     }
     throw new UnsupportedOperationException("Unsupported metric type: " + original.getType());
   }
-
-  /**
-   * A data class to store the result of a split operation performed on a single {@link MetricData}
-   * object.
-   */
-  private static class MetricDataSplitOperationResult {
-    private final Collection<Collection<MetricData>> preparedBatches;
-    private final BatchState lastInProgressBatch;
-
-    /**
-     * Creates a new MetricDataSplitOperationResult.
-     *
-     * @param preparedBatches The collection of prepared batches of metric data for export. Each
-     *     batch of {@link MetricData} objects is guaranteed to have at most {@link
-     *     #maxExportBatchSize} points.
-     * @param lastInProgressBatch The last batch that is still in progress. This batch may have less
-     *     than {@link #maxExportBatchSize} points.
-     */
-    MetricDataSplitOperationResult(
-        Collection<Collection<MetricData>> preparedBatches, BatchState lastInProgressBatch) {
-      this.preparedBatches = preparedBatches;
-      this.lastInProgressBatch = lastInProgressBatch;
-    }
-
-    Collection<Collection<MetricData>> getPreparedBatches() {
-      return preparedBatches;
-    }
-
-    BatchState getLastInProgressBatch() {
-      return lastInProgressBatch;
-    }
-  }
-
-  /**
-   * Tracks the active batch while batching stays linear: {@code metrics} is the current export
-   * payload being assembled and {@code points} is its running point count, so callers do not need
-   * to rescan the batch on every append.
-   */
-  private static final class BatchState {
-    private final Collection<MetricData> metrics;
-    private int points;
-
-    /**
-     * Creates the mutable state for the current in-progress batch.
-     *
-     * @param metrics metric entries collected into the current export batch
-     * @param points running total of data points across {@code metrics}
-     */
-    private BatchState(Collection<MetricData> metrics, int points) {
-      this.metrics = metrics;
-      this.points = points;
-    }
-  }
 }