Merge pull request #3732 from IBM/issue-3731-ra

Issue 3731 add reindex force option to fhir-bucket

Author: punktilious

.gitignore

@@ -39,6 +39,7 @@ fhirschema.log
 
 # Don't import pom backups
 **/pom.xml.versionsBackup
+dependency-reduced-pom.xml
 
 # Maven config
 .mvn/

fhir-bucket/README.md

@@ -48,13 +48,32 @@ The value for `--reindex-concurrent-requests` can be increased/decreased to maxi
 
 If the client-side-driven reindex is unable to be completed due to an error or timeout, the reindex can be resumed by using the `--reindex-start-with-index-id` parameter. If this needs to be done, first check the fhir-bucket log and find the first index ID that was not successful. Then, by specifying that index ID for the value of `--reindex-start-with-index-id` when starting the client-side-driven reindex, the reindex is resumed from that point, instead of starting completely over.
 
+When reindexing, the server compares a hash of the parameter values extracted from the resource with the value of the parameter hash last stored in the database. If the values are equal, the reindex operation skips further processing of the resource which can save a significant amount of time in cases where a search parameter configuration change affects only a subset of resources or resource types. However, some schema upgrades may change the way search parameters are stored and indexed. In such a case, use the `--reindex-force` option. When given, this instructs the reindex operation to ignore the parameter hash comparison and instead always store the new parameters. For example:
+
+```
+java \
+  -Djava.util.logging.config.file=logging.properties \
+  -jar "${JAR}" \
+  --fhir-properties your-fhir-server.properties \
+  --tenant-name your-tenant-name \
+  --max-concurrent-fhir-requests 100 \
+  --no-scan \
+  --reindex-tstamp 2020-12-01T00:00:00Z \
+  --reindex-resource-count 50 \
+  --reindex-concurrent-requests 20 \
+  --reindex-client-side-driven \
+  --reindex-force
+```
+
+
 | Property | Description |
 |---|---|
 | `--reindex-tstamp` | 'yyyy-MM-dd' or ISO 8601 dateTime which the reindex is using to start the operation|
 | `--reindex-resource-count` | The count to index in a single request. e.g. 100 with a maximum value is 1000.|
 | `--reindex-concurrent-requests` | The simultaneous requests used to execute concurrently |
 | `--reindex-client-side-driven` | Switches between Client driven $reindex and Server side driven reindex. True or false, false by default. |
 | `--reindex-start-with-index-id` | A index-id used with Client driven $reindex to drive reindexing from the Client side.|
+| `--reindex-force` | Force the reindex operation to replace the parameters, even if the parameter hash matches.|
 | `--fhir-properties` | Properties file for the IBM FHIR Server |
 
 ### IBM FHIR Server Properties - The FHIR Properties

fhir-bucket/src/main/java/com/ibm/fhir/bucket/app/Main.java

@@ -241,6 +241,9 @@ public class Main {
     // How many reindex calls should we run in parallel
     private int reindexConcurrentRequests = 1;
 
+    // Force reindex even if parameter hash matches
+    private boolean reindexForce = false;
+
     // The number of patients to fetch into the buffer
     private int patientBufferSize = 500000;
 
@@ -511,6 +514,9 @@ public void parseArgs(String[] args) {
                     throw new IllegalArgumentException("missing value for --reindex-resource-count");
                 }
                 break;
+            case "--reindex-force":
+                this.reindexForce = true;
+                break;
             case "--reindex-concurrent-requests":
                 if (i < args.length + 1) {
                     this.reindexConcurrentRequests = Integer.parseInt(args[++i]);
@@ -1136,9 +1142,9 @@ protected void scanAndLoad() {
      */
     private void doReindex() {
         if (this.clientSideDrivenReindex) {
-            this.driveReindexOperation = new ClientDrivenReindexOperation(fhirClient, reindexConcurrentRequests, reindexTstampParam, reindexResourceCount, reindexStartWithIndexId);
+            this.driveReindexOperation = new ClientDrivenReindexOperation(fhirClient, reindexConcurrentRequests, reindexTstampParam, reindexResourceCount, reindexStartWithIndexId, reindexForce);
         } else {
-            this.driveReindexOperation = new ServerDrivenReindexOperation(fhirClient, reindexConcurrentRequests, reindexTstampParam, reindexResourceCount);
+            this.driveReindexOperation = new ServerDrivenReindexOperation(fhirClient, reindexConcurrentRequests, reindexTstampParam, reindexResourceCount, reindexForce);
         }
         this.driveReindexOperation.init();
 

fhir-bucket/src/main/java/com/ibm/fhir/bucket/reindex/ClientDrivenReindexOperation.java

@@ -1,5 +1,5 @@
 /*
- * (C) Copyright IBM Corp. 2021
+ * (C) Copyright IBM Corp. 2021, 2022
  *
  * SPDX-License-Identifier: Apache-2.0
  */
@@ -44,6 +44,7 @@ public class ClientDrivenReindexOperation extends DriveReindexOperation {
     private static final String NOT_MODIFIED_AFTER_PARAM = "notModifiedAfter";
     private static final String AFTER_INDEX_ID_PARAM = "afterIndexId";
     private static final String INDEX_IDS_PARAM = "indexIds";
+    private static final String FORCE_PARAM = "force";
     private static final int MAX_RETRIEVE_COUNT = 1000;
     private static final int OFFER_TIMEOUT_IN_SEC = 30;
     private static final int POLL_TIMEOUT_IN_SEC = 5;
@@ -88,21 +89,26 @@ public class ClientDrivenReindexOperation extends DriveReindexOperation {
     // Number of threads currently running
     private AtomicInteger currentlyRunning = new AtomicInteger();
 
+    // Include the force parameter in the reindex operation request
+    private final boolean force;
+
     /**
      * Public constructor.
      * @param fhirClient the FHIR client
      * @param maxConcurrentRequests the number of threads to spin up
      * @param reindexTimestamp timestamp the reindex began
      * @param maxResourceCount resources processed per request per thread
      * @param startWithIndexId index ID from which to start, or null
+     * @param force force the reindex even if the parameter hash matches
      */
-    public ClientDrivenReindexOperation(FHIRBucketClient fhirClient, int maxConcurrentRequests, String reindexTimestamp, int maxResourceCount, String startWithIndexId) {
+    public ClientDrivenReindexOperation(FHIRBucketClient fhirClient, int maxConcurrentRequests, String reindexTimestamp, int maxResourceCount, String startWithIndexId, boolean force) {
         this.fhirClient = fhirClient;
         this.maxConcurrentRequests = maxConcurrentRequests;
         this.reindexTimestamp = reindexTimestamp;
         this.maxResourceCount = maxResourceCount;
         this.blockingQueue = new LinkedBlockingDeque<>(MAX_RETRIEVE_COUNT + (maxResourceCount * maxConcurrentRequests));
         this.inProgressIndexIds = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>(maxResourceCount * maxConcurrentRequests));
+        this.force = force;
         if (startWithIndexId != null) {
             // Subtract 1 since the $retrieve-index operation retrieves index IDs after a specified index ID
             this.lastIndexId = String.valueOf(Long.parseLong(startWithIndexId) - 1);
@@ -427,6 +433,7 @@ private boolean callReindexOperation(String indexIds) {
 
         Builder builder = Parameters.builder();
         builder.parameter(Parameter.builder().name(str(INDEX_IDS_PARAM)).value(str(indexIds)).build());
+        builder.parameter(Parameter.builder().name(str(FORCE_PARAM)).value(true).build());
         Parameters parameters = builder.build();
         String requestBody = FHIRBucketClientUtil.resourceToString(parameters);
 

fhir-bucket/src/main/java/com/ibm/fhir/bucket/reindex/ServerDrivenReindexOperation.java

@@ -1,5 +1,5 @@
 /*
- * (C) Copyright IBM Corp. 2020, 2021
+ * (C) Copyright IBM Corp. 2020, 2022
  *
  * SPDX-License-Identifier: Apache-2.0
  */
@@ -58,20 +58,28 @@ public class ServerDrivenReindexOperation extends DriveReindexOperation {
 
     /**
      * Public constructor
+     * 
      * @param client the FHIR client
      * @param maxConcurrentRequests the number of threads to spin up
+     * @param tstampParam
+     * @param resourceCountParam
+     * @param force
      */
-    public ServerDrivenReindexOperation(FHIRBucketClient fhirClient, int maxConcurrentRequests, String tstampParam, int resourceCountParam) {
+    public ServerDrivenReindexOperation(FHIRBucketClient fhirClient, int maxConcurrentRequests, String tstampParam, int resourceCountParam, boolean force) {
         this.fhirClient = fhirClient;
         this.maxConcurrentRequests = maxConcurrentRequests;
 
-        Parameters parameters = Parameters.builder()
+        Parameters.Builder builder = Parameters.builder()
                 .parameter(Parameter.builder().name(str("tstamp")).value(str(tstampParam)).build())
                 .parameter(Parameter.builder().name(str("resourceCount")).value(intValue(resourceCountParam)).build())
-                .build();
+                ;
+
+        if (force) {
+            builder.parameter(Parameter.builder().name(str("force")).value(true).build());
+        }
 
         // Serialize into the requestBody string used by all the threads
-        this.requestBody = FHIRBucketClientUtil.resourceToString(parameters);
+        this.requestBody = FHIRBucketClientUtil.resourceToString(builder.build());
 
         if (logger.isLoggable(Level.FINE)) {
             logger.fine("Reindex request parameters: " + requestBody);