-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #145 from kmwtechnology/LC-469
- Loading branch information
Showing
56 changed files
with
941 additions
and
83 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
80 changes: 80 additions & 0 deletions
80
lucille-core/src/main/java/com/kmwllc/lucille/core/RunnerManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| package com.kmwllc.lucille.core; | ||
|
|
||
| import com.kmwllc.lucille.core.Runner; | ||
| import com.kmwllc.lucille.core.Runner.RunType; | ||
| import com.typesafe.config.Config; | ||
| import com.typesafe.config.ConfigFactory; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.ExecutionException; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * Public API for starting lucille runs and viewing their status. Will be used by external resources, namely the Admin API to kick | ||
| * off lucille runs. | ||
| */ | ||
| public class RunnerManager { | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(RunnerManager.class); | ||
|
|
||
| // Use the eager-initialization singleton pattern | ||
| private static volatile RunnerManager instance = new RunnerManager(); | ||
|
|
||
| private RunnerManager() {} | ||
|
|
||
| public static RunnerManager getInstance() { | ||
| return instance; | ||
| } | ||
|
|
||
| // Check whether the CompleteableFuture exists and is not done | ||
| synchronized public boolean isRunning() { | ||
| return !future.isDone(); | ||
| } | ||
|
|
||
| /** | ||
| * Blocks the calling thread until the current lucille run is completed. This method is primarily intended to be used for testing, | ||
| * but has been made public to facilitate testing in other modules, namely lucille-plugins/lucille-api. | ||
| */ | ||
| public void waitForRunCompletion() throws ExecutionException, InterruptedException { | ||
| future.get(); | ||
| } | ||
|
|
||
| private CompletableFuture<Void> future = CompletableFuture.completedFuture(null); | ||
|
|
||
| /** | ||
| * Main entrypoint for kicking off lucille runs. This method spawns a new thread for lucille to run in, but will not start a new | ||
| * instance of lucille until the previous one terminates. | ||
| * | ||
| * @return boolean representing whether the lucille run was initiated or not. This will return false if and only if the lucille | ||
| * run was skipped due to the previous run still existing. | ||
| */ | ||
| synchronized public boolean run() { | ||
| return runWithConfig(ConfigFactory.load()); | ||
| } | ||
|
|
||
| /** | ||
| * Internal abstraction used to support testing. | ||
| */ | ||
| synchronized protected boolean runWithConfig(Config config) { | ||
| if (isRunning()) { | ||
| log.warn("Skipping new run; previous lucille run is still in progress."); | ||
| return false; | ||
| } | ||
|
|
||
| future = CompletableFuture.runAsync(() -> { | ||
| try { | ||
| log.info("Starting lucille run via the Runner Manager."); | ||
| log.info(config.entrySet().toString()); | ||
|
|
||
| // For now we will always use local mode without kafka | ||
| RunType runType = Runner.getRunType(false, true); | ||
|
|
||
| Runner.runWithResultLog(config, runType); | ||
| } catch (Exception e) { | ||
| log.error("Failed to run lucille via the Runner Manager.", e); | ||
| } | ||
| }); | ||
|
|
||
| return true; | ||
| } | ||
| } |
32 changes: 32 additions & 0 deletions
32
lucille-core/src/main/java/com/kmwllc/lucille/indexer/NopIndexer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| package com.kmwllc.lucille.indexer; | ||
|
|
||
| import com.kmwllc.lucille.core.Document; | ||
| import com.kmwllc.lucille.core.Indexer; | ||
| import com.kmwllc.lucille.message.IndexerMessenger; | ||
| import com.typesafe.config.Config; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * The NopIndexer performs no operations and does not send documents to any index. It is intended to be used for testing or | ||
| * validating that a pipeline is properly configured without ingesting content. | ||
| */ | ||
| public class NopIndexer extends Indexer { | ||
|
|
||
| public NopIndexer(Config config, IndexerMessenger messenger, boolean bypass, String metricsPrefix) { | ||
| super(config, messenger, metricsPrefix); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean validateConnection() { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| protected void sendToIndex(List<Document> documents) throws Exception { | ||
| // no-op | ||
| } | ||
|
|
||
| @Override | ||
| public void closeConnection() { | ||
| } | ||
| } |
24 changes: 24 additions & 0 deletions
24
lucille-core/src/test/java/com/kmwllc/lucille/connector/SleepConnector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| package com.kmwllc.lucille.connector; | ||
|
|
||
| import com.kmwllc.lucille.core.ConnectorException; | ||
| import com.kmwllc.lucille.core.Publisher; | ||
| import com.typesafe.config.Config; | ||
|
|
||
| public class SleepConnector extends AbstractConnector { | ||
|
|
||
| private final int duration; | ||
|
|
||
| public SleepConnector(Config config) { | ||
| super(config); | ||
| this.duration = config.getInt("duration"); | ||
| } | ||
|
|
||
| @Override | ||
| public void execute(Publisher publisher) throws ConnectorException { | ||
| try { | ||
| Thread.sleep(duration); | ||
| } catch (InterruptedException e) { | ||
| throw new ConnectorException("Sleep was interrupted", e); | ||
| } | ||
| } | ||
| } |
56 changes: 56 additions & 0 deletions
56
lucille-core/src/test/java/com/kmwllc/lucille/core/RunnerManagerTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package com.kmwllc.lucille.core; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| import com.typesafe.config.Config; | ||
| import com.typesafe.config.ConfigFactory; | ||
| import java.util.concurrent.ExecutionException; | ||
| import org.junit.Before; | ||
| import org.junit.Test; | ||
| import scala.Int; | ||
|
|
||
| public class RunnerManagerTest { | ||
|
|
||
| @Test | ||
| public void testRunnerManagerFull() throws InterruptedException, ExecutionException { | ||
| RunnerManager runnerManager = RunnerManager.getInstance(); | ||
| Config config = ConfigFactory.load("RunnerManagerTest/sleep.conf"); | ||
|
|
||
| // Ensure no lucille run is running at the start of the test | ||
| assertFalse(runnerManager.isRunning()); | ||
|
|
||
| // Kick off a lucille run and ensure it is not skipped | ||
| assertTrue(runnerManager.runWithConfig(config)); | ||
|
|
||
| // While we lucille is running, ensure lucille isRunning and a new run is skipped | ||
| assertTrue(runnerManager.isRunning()); | ||
| assertFalse(runnerManager.runWithConfig(config)); | ||
|
|
||
| // Wait until the run is over | ||
| runnerManager.waitForRunCompletion(); | ||
|
|
||
| // Ensure lucille is not running and make sure we can now kick off a new run | ||
| assertFalse(runnerManager.isRunning()); | ||
| assertTrue(runnerManager.runWithConfig(config)); | ||
|
|
||
| // Wait for all lucille threads to finish before exiting | ||
| runnerManager.waitForRunCompletion(); | ||
| } | ||
|
|
||
| @Test | ||
| public void testWaitForRunCompletion() throws InterruptedException, ExecutionException { | ||
| RunnerManager runnerManager = RunnerManager.getInstance(); | ||
| Config config = ConfigFactory.load("RunnerManagerTest/sleep.conf"); | ||
|
|
||
| // Ensure lucille is not running first | ||
| assertFalse(runnerManager.isRunning()); | ||
|
|
||
| runnerManager.runWithConfig(config); | ||
|
|
||
| // Ensure lucille is running, wait for it stop and ensure its stopped | ||
| assertTrue(runnerManager.isRunning()); | ||
| runnerManager.waitForRunCompletion(); | ||
| assertFalse(runnerManager.isRunning()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.