refactor: use JobProgress tracking in data value import [DHIS2-18885] (#19781)

* chore: use JobProgress tracking in data value import [DHIS2-18885]

* fix: dependencies and stage organisation

* fix: read ADX import directly using a reader adapter

* fix: test

* fix: pass NOOP progress tracker in tests

* fix: progress tracking of data import + bucketing of work items

* chore: sonar issues

* fix: AOC is a ADX group attribute

* fix: revert XML export format changes as test fail that are hard to adjust

Author: jbee

dhis-2/dhis-api/src/main/java/org/hisp/dhis/scheduling/JobProgress.java

@@ -276,6 +276,14 @@ default void failedStage(@Nonnull Exception cause) {
     failedStage(getMessage(cause));
   }
 
+  /**
+   * @since 2.42
+   * @param size number of work items to join into one entry
+   */
+  default void setWorkItemBucketing(int size) {
+    // by default this is not supported and no bucketing will occur
+  }
+
   default void startingWorkItem(@Nonnull String description, Object... args) {
     startingWorkItem(format(description, args), FailurePolicy.PARENT);
   }

dhis-2/dhis-services/dhis-service-core/src/main/java/org/hisp/dhis/scheduling/RecordingJobProgress.java

@@ -41,6 +41,8 @@
 import lombok.extern.slf4j.Slf4j;
 import org.hisp.dhis.feedback.ErrorCode;
 import org.hisp.dhis.message.MessageService;
+import org.hisp.dhis.system.notification.NotificationLevel;
+import org.hisp.dhis.system.notification.Notifier;
 import org.hisp.dhis.tracker.imports.validation.ValidationCode;
 import org.hisp.dhis.user.CurrentUserUtil;
 import org.hisp.dhis.user.UserDetails;
@@ -66,7 +68,15 @@ public class RecordingJobProgress implements JobProgress {
    *     recoding objects
    */
   public static JobProgress transitory() {
-    return new RecordingJobProgress(null, null, JobProgress.noop(), true, () -> {}, false, true);
+    return transitory(null, null);
+  }
+
+  public static JobProgress transitory(JobConfiguration job, Notifier notifier) {
+    JobProgress track =
+        notifier == null
+            ? JobProgress.noop()
+            : new NotifierJobProgress(notifier, job, NotificationLevel.INFO);
+    return new RecordingJobProgress(null, null, track, true, () -> {}, false, true);
   }
 
   @CheckForNull private final MessageService messageService;
@@ -87,6 +97,9 @@ public static JobProgress transitory() {
   private final ThreadLocal<Item> incompleteItem = new ThreadLocal<>();
   private final boolean usingErrorNotification;
 
+  private int bucketingSize;
+  private int bucketed;
+
   public RecordingJobProgress(JobConfiguration configuration) {
     this(null, configuration, JobProgress.noop(), true, () -> {}, false, false);
   }
@@ -295,6 +308,8 @@ public void startingStage(
     skipCurrentStage.set(false);
     tracker.startingStage(description, workItems);
     incompleteItem.remove();
+    bucketingSize = 1;
+    bucketed = 0;
     Stage stage =
         addStageRecord(getOrAddLastIncompleteProcess(), description, workItems, onFailure);
     logInfo(stage, "", description);
@@ -315,6 +330,7 @@ public void completedStage(String summary, Object... args) {
     String message = format(summary, args);
     tracker.completedStage(message);
     Stage stage = getOrAddLastIncompleteStage();
+    autoCompleteWorkItemBucket();
     stage.complete(message);
     logInfo(stage, "completed", message);
   }
@@ -326,6 +342,7 @@ public void failedStage(@Nonnull String error, Object... args) {
     String message = format(error, args);
     tracker.failedStage(message);
     Stage stage = getOrAddLastIncompleteStage();
+    autoCompleteWorkItemBucket();
     stage.completeExceptionally(message, null);
     if (stage.getOnFailure() != FailurePolicy.SKIP_STAGE) {
       automaticAbort(message, null);
@@ -341,6 +358,7 @@ public void failedStage(@Nonnull Exception cause) {
     tracker.failedStage(cause);
     String message = getMessage(cause);
     Stage stage = getOrAddLastIncompleteStage();
+    autoCompleteWorkItemBucket();
     stage.completeExceptionally(message, cause);
     if (stage.getOnFailure() != FailurePolicy.SKIP_STAGE) {
       automaticAbort(message, cause);
@@ -349,28 +367,57 @@ public void failedStage(@Nonnull Exception cause) {
     logError(stage, cause, message);
   }
 
+  @Override
+  public void setWorkItemBucketing(int size) {
+    bucketingSize = Math.max(1, size);
+  }
+
   @Override
   public void startingWorkItem(@Nonnull String description, @Nonnull FailurePolicy onFailure) {
-    observer.run();
+    if (bucketed % bucketingSize == 0) {
+      observer.run();
 
-    tracker.startingWorkItem(description, onFailure);
-    Item item = addItemRecord(getOrAddLastIncompleteStage(), description, onFailure);
-    logDebug(item, "started", description);
+      tracker.startingWorkItem(description, onFailure);
+      Item item = addItemRecord(getOrAddLastIncompleteStage(), description, onFailure);
+      logDebug(item, "started", description);
+    }
+    bucketed++;
   }
 
   @Override
   public void completedWorkItem(String summary, Object... args) {
+    if (bucketed % bucketingSize == 0) {
+      completeWorkItemBucket(summary, args);
+    }
+  }
+
+  private void completeWorkItemBucket(String summary, Object... args) {
     observer.run();
 
-    String message = format(summary, args);
-    tracker.completedWorkItem(message);
     Item item = getOrAddLastIncompleteItem();
+    String message =
+        summary == null && bucketingSize > 1 ? getBucketSummary(item) : format(summary, args);
+    tracker.completedWorkItem(message);
     item.complete(message);
     logDebug(item, "completed", message);
   }
 
+  @Nonnull
+  private String getBucketSummary(Item item) {
+    int n = bucketed % bucketingSize;
+    return item.getDescription() + " +" + (n == 0 ? bucketingSize : n);
+  }
+
+  private void autoCompleteWorkItemBucket() {
+    if (bucketingSize > 1) {
+      Item item = incompleteItem.get();
+      if (item != null && !item.isComplete()) completeWorkItemBucket(null);
+    }
+  }
+
   @Override
   public void failedWorkItem(@Nonnull String error, Object... args) {
+    bucketed = 0; // reset to restart bucketing on next item
     observer.run();
 
     String message = format(error, args);
@@ -385,6 +432,7 @@ public void failedWorkItem(@Nonnull String error, Object... args) {
 
   @Override
   public void failedWorkItem(@Nonnull Exception cause) {
+    bucketed = 0; // reset to restart bucketing on next item
     observer.run();
 
     tracker.failedWorkItem(cause);

dhis-2/dhis-services/dhis-service-dxf2/pom.xml

@@ -99,10 +99,6 @@
       <groupId>org.springframework</groupId>
       <artifactId>spring-web</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.springframework.security</groupId>
-      <artifactId>spring-security-core</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.hibernate</groupId>
       <artifactId>hibernate-core-jakarta</artifactId>

dhis-2/dhis-services/dhis-service-dxf2/src/main/java/org/hisp/dhis/dxf2/adx/AdxDataService.java

@@ -29,11 +29,12 @@
 
 import java.io.InputStream;
 import java.io.OutputStream;
+import javax.annotation.Nonnull;
 import org.hisp.dhis.datavalue.DataExportParams;
 import org.hisp.dhis.dxf2.common.ImportOptions;
 import org.hisp.dhis.dxf2.datavalueset.DataValueSetQueryParams;
 import org.hisp.dhis.dxf2.importsummary.ImportSummary;
-import org.hisp.dhis.scheduling.JobConfiguration;
+import org.hisp.dhis.scheduling.JobProgress;
 
 /**
  * @author bobj
@@ -84,10 +85,11 @@ public interface AdxDataService {
    *
    * @param in the InputStream.
    * @param importOptions the importOptions.
-   * @param id the task id, can be null.
+   * @param progress to track progress
    * @return an ImportSummaries collection of ImportSummary for each DataValueSet.
    */
-  ImportSummary saveDataValueSet(InputStream in, ImportOptions importOptions, JobConfiguration id);
+  ImportSummary saveDataValueSet(
+      InputStream in, ImportOptions importOptions, @Nonnull JobProgress progress);
 
   /**
    * Get data. Writes adx export data to output stream.

dhis-2/dhis-services/dhis-service-dxf2/src/main/java/org/hisp/dhis/dxf2/adx/AdxDataValueSetReader.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright (c) 2004-2025, University of Oslo
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ * Redistributions of source code must retain the above copyright notice, this
+ * list of conditions and the following disclaimer.
+ *
+ * Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * Neither the name of the HISP project nor the names of its contributors may
+ * be used to endorse or promote products derived from this software without
+ * specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package org.hisp.dhis.dxf2.adx;
+
+import static java.lang.Boolean.parseBoolean;
+
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import lombok.RequiredArgsConstructor;
+import org.hisp.dhis.dxf2.datavalueset.DataValueEntry;
+import org.hisp.dhis.dxf2.datavalueset.DataValueSet;
+import org.hisp.dhis.dxf2.datavalueset.DataValueSetReader;
+import org.hisp.staxwax.reader.XMLReader;
+
+@RequiredArgsConstructor
+public class AdxDataValueSetReader implements DataValueSetReader, DataValueEntry {
+
+  private final XMLReader adxReader;
+  private final Consumer<Map<String, String>> handleGroup;
+  private final Predicate<Map<String, String>> handleValue;
+
+  private Map<String, String> groupAttributes;
+  private Map<String, String> valueAttributes;
+
+  @Override
+  public DataValueSet readHeader() {
+    adxReader.moveToStartElement(AdxDataService.ROOT, AdxDataService.NAMESPACE);
+    return new DataValueSet();
+  }
+
+  @Override
+  public DataValueEntry readNext() {
+    if (groupAttributes == null) {
+      if (!adxReader.moveToStartElement(AdxDataService.GROUP, AdxDataService.NAMESPACE))
+        return null;
+      groupAttributes = adxReader.readAttributes();
+      handleGroup.accept(groupAttributes);
+    }
+    if (!adxReader.moveToStartElement(AdxDataService.DATAVALUE, AdxDataService.GROUP)) {
+      groupAttributes = null;
+      // end of the group, try next by recursively calling itself
+      return readNext();
+    }
+    valueAttributes = adxReader.readAttributes();
+    if (handleValue.test(valueAttributes)) {
+      // if data element type is not numeric we need to pick out the
+      // 'annotation' element
+      adxReader.moveToStartElement(AdxDataService.ANNOTATION, AdxDataService.DATAVALUE);
+      if (adxReader.isStartElement(AdxDataService.ANNOTATION)) {
+        valueAttributes.put(AdxDataService.VALUE, adxReader.getElementValue());
+      } else {
+        throw new IllegalArgumentException("DataElement expects text annotation");
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public void close() {
+    adxReader.closeReader();
+  }
+
+  @Override
+  public String getDataElement() {
+    return valueAttributes.get(AdxDataService.DATAELEMENT);
+  }
+
+  @Override
+  public String getPeriod() {
+    return groupAttributes.get(AdxDataService.PERIOD);
+  }
+
+  @Override
+  public String getOrgUnit() {
+    return groupAttributes.get(AdxDataService.ORGUNIT);
+  }
+
+  @Override
+  public String getCategoryOptionCombo() {
+    return valueAttributes.get(AdxDataService.CATOPTCOMBO);
+  }
+
+  @Override
+  public String getAttributeOptionCombo() {
+    return groupAttributes.get(AdxDataService.ATTOPTCOMBO);
+  }
+
+  @Override
+  public String getValue() {
+    return valueAttributes.get(AdxDataService.VALUE);
+  }
+
+  @Override
+  public String getStoredBy() {
+    return valueAttributes.get("storedBy");
+  }
+
+  @Override
+  public String getCreated() {
+    return valueAttributes.get("created");
+  }
+
+  @Override
+  public String getLastUpdated() {
+    return valueAttributes.get("lastUpdated");
+  }
+
+  @Override
+  public String getComment() {
+    return valueAttributes.get("comment");
+  }
+
+  @Override
+  public boolean getFollowup() {
+    return parseBoolean(valueAttributes.get("followup"));
+  }
+
+  @Override
+  public Boolean getDeleted() {
+    String deleted = valueAttributes.get("deleted");
+    return deleted == null ? null : parseBoolean(deleted);
+  }
+}