Add isReady() and document lazy model loading in the Java library (#17)

README.md

@@ -657,16 +657,18 @@ BidRequestEvaluatorFactory bidRequestEvaluatorFactory =
         bucket,
         executor);
 
-// initialize background tasks to periodically fetch new signal data to evaluate.
+// load the configurations and schedule background tasks to periodically fetch new signal data to evaluate.
 bidRequestEvaluatorFactory.getTaskInitializer().init();
 
 // create an evaluator instance to receive recommendations. 
 BidRequestEvaluator bidRequestEvaluator = bidRequestEvaluatorFactory.getEvaluator();
 
-// sleep to ensure initialization is complete before evaluating requests.
-Thread.sleep(10000);
+// report readiness (e.g. from a health check) instead of sleeping for a fixed time.
+boolean ready = bidRequestEvaluatorFactory.isReady();
 ```
 
+`init()` loads the model and experiment configurations and schedules the periodic refresh; it throws if the configurations cannot be loaded within `task.overall.execution-timeout`. It does not wait for the model results: each model loading task runs first after a random delay of up to one refresh period (5 minutes by default), so that hosts that start together do not all read S3 at once. Until a model's results are loaded, that model returns its no-match value: 1.0 (forward) for a `LowValue` model and 0.0 (low value) for a `HighValue` model, combined with the other models by the aggregation as usual. With only `LowValue` models, evaluations therefore fail open (traffic is forwarded) until the models load. `isReady()` turns true once `init()` has completed, the feature registries are populated and every model loading task has finished its first run, which can take up to one refresh period (the longest configured model-result period) plus the model load time. A first run that fails, for example because a model file is missing or S3 returns an error, still counts as finished: that model keeps returning its no-match value until a later refresh loads it. Once true, `isReady()` stays true. Poll it from a readiness probe, or before taking traffic, if you prefer not to evaluate before the models are loaded.
+
 In addition, update the logging configurations under `src/main/resources/log4j2.xml` as needed.
 
 For the Golang library, the initialization method is slightly different.

java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/BidRequestEvaluatorFactory.java

@@ -32,4 +32,45 @@ public static BidRequestEvaluatorFactory create(String supplierName, AwsCredenti
 
     public abstract BidRequestEvaluator getEvaluator();
 
+    /**
+     * Tells whether the evaluator has had its first chance to load the data it needs to make recommendations.
+     * <p>
+     * True once all of the following hold:
+     * </p>
+     * <ul>
+     *   <li>{@code init()} has completed normally on a {@link TaskInitializer} obtained from this factory, so the
+     *       model and experiment configurations are loaded;</li>
+     *   <li>every feature extractor and transformer is registered, which happens within about one second of
+     *       {@code init()} starting, longer if the executor has no free thread;</li>
+     *   <li>every model-result and bloom-filter loading task created for that {@link TaskInitializer} has finished
+     *       its first periodic run.</li>
+     * </ul>
+     * <p>
+     * The first periodic run starts after a random delay of up to one refresh period (300 s by default; the
+     * longest of the configured model-result periods), so this can take up to one refresh period plus the model
+     * load time after {@code init()} returns, longer if the executor has no free thread. Once true it stays true;
+     * later refresh failures do not reset it. It stays false if {@code init()} throws, and while a feature
+     * extractor or transformer provider has not been registered, for example because it failed to load.
+     * </p>
+     * <p>
+     * A first run counts as finished whatever its outcome: a model file that is missing, an S3 error or a loader
+     * exception does not block readiness.
+     * </p>
+     * <p>
+     * Until a model's results are loaded, evaluations still run: that model returns its no-match value, which
+     * forwards for a {@code LowValue} model (1.0) and marks the request low value for a {@code HighValue} model
+     * (0.0), and the aggregation combines it with the other models as usual. With only {@code LowValue} models,
+     * evaluations therefore fail open (forward) until the models load.
+     * </p>
+     * <p>
+     * This default implementation throws; {@link DefaultBidRequestEvaluatorFactory} overrides it.
+     * </p>
+     *
+     * @return true when the conditions above hold.
+     * @throws UnsupportedOperationException if the factory does not track readiness.
+     */
+    public boolean isReady() {
+        throw new UnsupportedOperationException("isReady is not implemented by " + getClass().getName());
+    }
+
 }

java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/DefaultBidRequestEvaluatorFactory.java

@@ -14,7 +14,9 @@
 import com.amazon.demanddriventrafficevaluator.evaluation.experiment.ExperimentManager;
 import com.amazon.demanddriventrafficevaluator.modelfeature.Extraction;
 import com.amazon.demanddriventrafficevaluator.modelfeature.Transformation;
+import com.amazon.demanddriventrafficevaluator.modelfeature.extractor.Extractor;
 import com.amazon.demanddriventrafficevaluator.modelfeature.extractor.ExtractorRegistry;
+import com.amazon.demanddriventrafficevaluator.modelfeature.transformer.Transformer;
 import com.amazon.demanddriventrafficevaluator.modelfeature.transformer.TransformerRegistry;
 import com.amazon.demanddriventrafficevaluator.repository.dao.BloomFilterDao;
 import com.amazon.demanddriventrafficevaluator.repository.dao.Dao;
@@ -33,9 +35,11 @@
 import com.amazon.demanddriventrafficevaluator.repository.provider.model.RuleBasedModelResultProvider;
 import com.amazon.demanddriventrafficevaluator.task.InitializerTask;
 import com.amazon.demanddriventrafficevaluator.task.InitializerTaskOnPeriodicTask;
+import com.amazon.demanddriventrafficevaluator.task.PeriodicTaskWithRandomizedStart;
 import com.amazon.demanddriventrafficevaluator.task.TaskConfiguration;
 import com.amazon.demanddriventrafficevaluator.task.TaskInitializer;
 import com.amazon.demanddriventrafficevaluator.task.dataloading.BloomFilterPeriodicLoadingTask;
+import com.amazon.demanddriventrafficevaluator.task.dataloading.ModelResultPeriodicLoadingInitializerTask;
 import com.amazon.demanddriventrafficevaluator.util.PropertiesUtil;
 import lombok.extern.log4j.Log4j2;
 import org.apache.commons.configuration2.Configuration;
@@ -44,7 +48,11 @@
 import java.io.InputStream;
 import java.util.List;
 import java.util.Map;
+import java.util.ServiceConfigurationError;
+import java.util.ServiceLoader;
+import java.util.Set;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.stream.Collectors;
 
 /**
  * A factory for creating BidRequestEvaluator instances that support both rule-based
@@ -75,6 +83,12 @@ public class DefaultBidRequestEvaluatorFactory extends BidRequestEvaluatorFactor
     private final String bucket;
     private final ScheduledThreadPoolExecutor executor;
     private final Dao<String, InputStream> fileDao;
+    // The model loading tasks of the last TaskInitializer whose init() completed; null until one has.
+    private volatile List<PeriodicTaskWithRandomizedStart> modelLoadingTasks;
+    private volatile boolean ready;
+    // The names the registry setup tasks register; listed once, on the first isReady() call that needs them.
+    private volatile Set<String> extractorNames;
+    private volatile Set<String> transformerNames;
 
     public DefaultBidRequestEvaluatorFactory(
             String supplierName,
@@ -110,9 +124,53 @@ public TaskInitializer getTaskInitializer() {
         }
 
         List<InitializerTask> stageOneTasks = getStageOneTasks();
-        List<InitializerTask> stageTwoTasks = getStageTwoTasks();
+        ModelResultPeriodicLoadingInitializerTask ruleBasedTask =
+                defaultTaskInitializerFactory.getInitializerTaskForPeriodicLoadingRuleBasedModelResult();
+        InitializerTaskOnPeriodicTask bloomFilterTask = getInitializerTaskForPeriodicLoadingBloomFilterModelResult();
+        List<InitializerTask> stageTwoTasks = List.of(ruleBasedTask, bloomFilterTask);
+        List<PeriodicTaskWithRandomizedStart> tasks = List.of(ruleBasedTask.getTask(), bloomFilterTask.getTask());
 
-        return new TaskInitializer(stageOneTasks, stageTwoTasks, overallTimeoutMs);
+        return new TaskInitializer(stageOneTasks, stageTwoTasks, overallTimeoutMs, () -> modelLoadingTasks = tasks);
+    }
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * This factory waits for the rule-based model-result and the bloom-filter loading tasks of the
+     * {@link TaskInitializer} whose {@code init()} completed last.
+     * </p>
+     */
+    @Override
+    public boolean isReady() {
+        if (ready) {
+            return true;
+        }
+        List<PeriodicTaskWithRandomizedStart> tasks = modelLoadingTasks;
+        if (tasks == null || !tasks.stream().allMatch(PeriodicTaskWithRandomizedStart::hasFinishedFirstRun)) {
+            return false;
+        }
+        try {
+            if (extractorNames == null || transformerNames == null) {
+                extractorNames = getProviderNames(Extractor.class);
+                transformerNames = getProviderNames(Transformer.class);
+            }
+        } catch (ServiceConfigurationError e) {
+            log.error("isReady: cannot list the feature extractor and transformer providers", e);
+            return false;
+        }
+        if (ExtractorRegistryFactory.getInstance().getSingleton().getRecords().keySet().containsAll(extractorNames)
+                && TransformerRegistryFactory.getInstance().getSingleton().getRecords().keySet()
+                .containsAll(transformerNames)) {
+            ready = true;
+        }
+        return ready;
+    }
+
+    // The names the registry setup tasks register: the simple class names of the ServiceLoader providers.
+    private static Set<String> getProviderNames(Class<?> type) {
+        return ServiceLoader.load(type).stream()
+                .map(provider -> provider.type().getSimpleName())
+                .collect(Collectors.toUnmodifiableSet());
     }
 
     @Override
@@ -167,13 +225,7 @@ private List<InitializerTask> getStageOneTasks() {
         return defaultTaskInitializerFactory.getStageOneTasks();
     }
 
-    private List<InitializerTask> getStageTwoTasks() {
-        InitializerTask ruleBasedTask = defaultTaskInitializerFactory.getInitializerTaskForPeriodicLoadingRuleBasedModelResult();
-        InitializerTask bloomFilterTask = getInitializerTaskForPeriodicLoadingBloomFilterModelResult();
-        return List.of(ruleBasedTask, bloomFilterTask);
-    }
-
-    private InitializerTask getInitializerTaskForPeriodicLoadingBloomFilterModelResult() {
+    private InitializerTaskOnPeriodicTask getInitializerTaskForPeriodicLoadingBloomFilterModelResult() {
         LocalCacheRegistry localCacheRegistry = DefaultLocalCacheRegistryFactory.getInstance().getDefaultLocalCacheRegistrySingleton();
         LocalCacheDao<String, ModelConfiguration> modelConfigurationCacheDao = new LocalCacheDao<>(localCacheRegistry);
         ConfigurationProvider<ModelConfiguration> modelConfigurationProvider = new ModelConfigurationProvider(modelConfigurationCacheDao);

java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/DefaultTaskInitializerFactory.java

@@ -165,7 +165,7 @@ private InitializerTaskOnPeriodicTask getInitializerTaskForPeriodicLoadingExperi
         );
     }
 
-    InitializerTask getInitializerTaskForPeriodicLoadingRuleBasedModelResult() {
+    ModelResultPeriodicLoadingInitializerTask getInitializerTaskForPeriodicLoadingRuleBasedModelResult() {
         LocalCacheDao<String, ModelConfiguration> modelConfigurationCacheDao = new LocalCacheDao<>(localCacheRegistry);
         ConfigurationProvider<ModelConfiguration> modelConfigurationProvider = new ModelConfigurationProvider(modelConfigurationCacheDao);
         Dao<String, Double> modelResultsCacheDao = new LocalCacheDao<>(localCacheRegistry);
@@ -216,7 +216,7 @@ private InitializerTaskOnPeriodicTask getInitializerTaskForPeriodicLoadingConfig
         );
     }
 
-    private InitializerTask getInitializerTaskForPeriodicLoadingModelResult(
+    private ModelResultPeriodicLoadingInitializerTask getInitializerTaskForPeriodicLoadingModelResult(
             String taskName,
             long periodMs,
             int maximumAttempts,

java/src/main/java/com/amazon/demanddriventrafficevaluator/task/InitializerTaskOnPeriodicTask.java

@@ -18,6 +18,13 @@ public InitializerTaskOnPeriodicTask(
         this.task = task;
     }
 
+    /**
+     * @return the periodic task that {@link #run()} initializes.
+     */
+    public PeriodicTaskWithRandomizedStart getTask() {
+        return task;
+    }
+
     /**
      * When an object implementing interface {@code Runnable} is used
      * to create a thread, starting the thread causes the object's

java/src/main/java/com/amazon/demanddriventrafficevaluator/task/PeriodicTaskWithRandomizedStart.java

@@ -3,6 +3,7 @@
 
 package com.amazon.demanddriventrafficevaluator.task;
 
+import lombok.AccessLevel;
 import lombok.Getter;
 
 import java.util.Random;
@@ -30,6 +31,10 @@ public abstract class PeriodicTaskWithRandomizedStart {
 
     private final Random random = new Random(System.currentTimeMillis());
 
+    // Written on the scheduler thread, read by callers of hasFinishedFirstRun().
+    @Getter(AccessLevel.NONE)
+    private volatile boolean firstRunFinished;
+
     protected PeriodicTaskWithRandomizedStart(String sspIdentifier, String taskName, long periodMs, ScheduledThreadPoolExecutor executor) {
         this.sspIdentifier = sspIdentifier;
         this.taskName = taskName;
@@ -49,6 +54,20 @@ protected PeriodicTaskWithRandomizedStart(String sspIdentifier, String taskName,
      */
     abstract public void initialize();
 
+    /**
+     * Tells whether the first scheduled execution of this task has finished.
+     * <p>
+     * True once the first execution scheduled by {@link #schedulePeriodically(long)} has returned or thrown,
+     * whatever its outcome. An execution run directly through {@link #executeTask()}, for example by
+     * {@link #initialize()}, does not count. Once true it stays true.
+     * </p>
+     *
+     * @return true when the first scheduled execution has finished.
+     */
+    public boolean hasFinishedFirstRun() {
+        return firstRunFinished;
+    }
+
     /**
      * Schedules the task to run periodically with a specified initial delay.
      *
@@ -61,6 +80,8 @@ protected void schedulePeriodically(long initialDelay) {
             } catch (Exception e) {
                 log.error("{}: task failed", taskName, e);
                 // Optionally rethrow or handle the exception
+            } finally {
+                firstRunFinished = true;
             }
         }, initialDelay, periodMs, TimeUnit.MILLISECONDS);
     }

java/src/main/java/com/amazon/demanddriventrafficevaluator/task/TaskInitializer.java

@@ -4,6 +4,7 @@
 package com.amazon.demanddriventrafficevaluator.task;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CompletionService;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorCompletionService;
@@ -25,13 +26,24 @@ public class TaskInitializer {
     private final List<InitializerTask> stageOneTasks;
     private final List<InitializerTask> stageTwoTasks;
     private final long overallTimeoutMs;
+    private final Runnable onInitialized;
     private ExecutorService executorPool;
     private long taskElapsedTime = 0L;
 
     public TaskInitializer(List<InitializerTask> stageOneTasks, List<InitializerTask> stageTwoTasks, long overallTimeoutMs) {
+        this(stageOneTasks, stageTwoTasks, overallTimeoutMs, () -> { });
+    }
+
+    /**
+     * @param onInitialized Run by {@link #init()} once the tasks of both stages have completed without an exception;
+     *                      not run when there are no stage one tasks.
+     */
+    public TaskInitializer(List<InitializerTask> stageOneTasks, List<InitializerTask> stageTwoTasks, long overallTimeoutMs,
+                           Runnable onInitialized) {
         this.stageOneTasks = stageOneTasks;
         this.stageTwoTasks = stageTwoTasks;
         this.overallTimeoutMs = overallTimeoutMs;
+        this.onInitialized = Objects.requireNonNull(onInitialized, "onInitialized");
     }
 
     /**
@@ -57,6 +69,18 @@ static long calculateDelay(int timesAlreadyAttempted, final long currentDelay, l
      * This method executes stage one tasks, followed by stage two tasks if present.
      * It manages the execution pool and handles overall timeout.
      * </p>
+     * <p>
+     * With the default factories, stage one loads the model and experiment configurations and
+     * schedules the feature extractor and transformer registration, which runs within about one
+     * second; stage two only schedules the periodic model-result and bloom-filter loading. The first
+     * model load runs on that schedule after a random delay of up to one refresh period (300 s by
+     * default), so this method does not wait for the model results. Until they are loaded, each
+     * model returns its no-match value: 1.0 (forward) for a {@code LowValue} model and 0.0 (low
+     * value) for a {@code HighValue} model. Callers that need to know when the first load has
+     * finished use {@code BidRequestEvaluatorFactory.isReady()}.
+     * </p>
+     *
+     * @throws RuntimeException if a task fails after its retries or the overall timeout elapses.
      */
     public void init() {
         if (stageOneTasks != null && !stageOneTasks.isEmpty()) {
@@ -65,6 +89,7 @@ public void init() {
                 if (stageTwoTasks != null && !stageTwoTasks.isEmpty()) {
                     executeTasks(stageTwoTasks, "StageTwo", System.currentTimeMillis());
                 }
+                onInitialized.run();
             } finally {
                 executorPool.shutdown();
                 try {
@@ -117,6 +142,10 @@ private void tasksResultIterator(CompletionService<Void> completionService, List
             }
             taskElapsedTime += elapsed;
         }
+        if (executedTaskCount < tasks.size()) {
+            throw new RuntimeException("Timed out after " + executedTaskCount + " of " + tasks.size() + " tasks and "
+                    + (System.currentTimeMillis() - startTime) + " ms");
+        }
     }
 
     /**

java/src/main/java/com/amazon/demanddriventrafficevaluator/task/dataloading/BloomFilterPeriodicLoadingTask.java

@@ -95,7 +95,8 @@ public void executeTask() {
      * <p>
      * This method sets up the periodic execution schedule for the task.
      * Unlike some other tasks, this does not perform an immediate execution
-     * upon initialization.
+     * upon initialization: the first execution runs after a random delay of
+     * up to one period, and {@link #hasFinishedFirstRun()} tells when it has finished.
      * </p>
      */
     @Override

java/src/main/java/com/amazon/demanddriventrafficevaluator/task/dataloading/ModelResultPeriodicLoadingInitializerTask.java

@@ -14,6 +14,13 @@ public ModelResultPeriodicLoadingInitializerTask(String taskName, int maximumAtt
         this.task = task;
     }
 
+    /**
+     * @return the periodic task that {@link #run()} initializes.
+     */
+    public ModelResultPeriodicLoadingTask getTask() {
+        return task;
+    }
+
     /**
      * When an object implementing interface {@code Runnable} is used
      * to create a thread, starting the thread causes the object's

java/src/main/java/com/amazon/demanddriventrafficevaluator/task/dataloading/ModelResultPeriodicLoadingTask.java

@@ -86,7 +86,8 @@ public void executeTask() {
      * <p>
      * This method sets up the periodic execution schedule for the task.
      * Unlike some other tasks, this does not perform an immediate execution
-     * upon initialization.
+     * upon initialization: the first execution runs after a random delay of
+     * up to one period, and {@link #hasFinishedFirstRun()} tells when it has finished.
      * </p>
      */
     @Override