first commit

src/main/java/fr/natan/akkastreamfileprocessingapi/controller/TvSeriesController.java

 import org.springframework.web.bind.annotation.RestController;
 import play.api.libs.json.JsValue;
 import play.api.libs.json.Json;
+import scala.Option;
 import scala.collection.IndexedSeq;
 import scala.concurrent.Future;
 
 
     @RequestMapping(value = "/persons/id/{personID}", method = RequestMethod.GET)
     private String getPersonByID(@PathVariable(name = "personID") String personID) throws ExecutionException, InterruptedException {
-        Future<ModelsAndJsonMap.Person> futurePerson = akkaStreamFilesProcessing.getPersonByIdFuture(personID);
+        Future<Option<ModelsAndJsonMap.Person>> futurePerson = akkaStreamFilesProcessing.getPersonByIdFuture(personID);
         CompletableFuture<ModelsAndJsonMap.Person> completableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(futurePerson);
         ModelsAndJsonMap.Person person = completableFuture.get();
     private String getPersonByName(@PathVariable(name = "primaryName") String primaryName)
             throws ExecutionException, InterruptedException {
 
-        Future<ModelsAndJsonMap.Person> personFuture = akkaStreamFilesProcessing.getPersonByNameFuture(primaryName);
+        Future<Option<ModelsAndJsonMap.Person>> personFuture = akkaStreamFilesProcessing.getPersonByNameFuture(primaryName);
         CompletableFuture<ModelsAndJsonMap.Person> personCompletableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(personFuture);
         ModelsAndJsonMap.Person person = personCompletableFuture.get();
     @RequestMapping(value = "/tvseries/id/{tvSerieId}", method = RequestMethod.GET)
     private String getTvSerieByID(@PathVariable(name = "tvSerieId") String tvSerieID)
             throws ExecutionException, InterruptedException {
-        Future<ModelsAndJsonMap.TvSerie> tvSerieFuture = akkaStreamFilesProcessing.getTvSerieByIdFuture(tvSerieID);
+        Future<Option<ModelsAndJsonMap.TvSerie>> tvSerieFuture = akkaStreamFilesProcessing.getTvSerieByIdFuture(tvSerieID);
         CompletableFuture<ModelsAndJsonMap.TvSerie> tvSerieCompletableFuture = CompletableFutureBuilder
                 .buildcompletableFuture1(tvSerieFuture);
 

src/main/java/fr/natan/akkastreamfileprocessingapi/futurecompleteness/CompletableFutureBuilder.java

 package fr.natan.akkastreamfileprocessingapi.futurecompleteness;
 
+import scala.Option;
 import scala.collection.IndexedSeq;
 import scala.concurrent.Future;
 
 public class CompletableFutureBuilder<T> {
 
    private static final Logger logger = Logger.getLogger("log");
-    public static<T> CompletableFuture<T> buildcompletableFuture1(Future<T> futureT) {
+    public static<T> CompletableFuture<T> buildcompletableFuture1(Future<Option<T>> futureT) {
         CompletableFuture<T> completableFuture = new CompletableFuture<>();
 
 
                     throw new RuntimeException(e);
                 }
             }
-            T task = futureT.value().get().get();
+            T task = futureT.value().get().get().get();
             completableFuture.complete(task);
         });
 

src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamFileProcessingFuture.scala

 import scala.concurrent.Future
 
 trait AkkaStreamFileProcessingFuture {
-  def getPersonByIdFuture(personID: String): Future[Person]
-  def getPersonByNameFuture(primaryName: String):  Future[Person]
+  def getPersonByIdFuture(personID: String): Future[Option[Person]]
+  def getPersonByNameFuture(primaryName: String):  Future[Option[Person]]
 
-  def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie]
+  def getTvSerieByIdFuture(tvSerieID: String): Future[Option[TvSerie]]
   def getTvSeriesByPrimaryTitleFuture(tvSerieTitle: String):Future[IndexedSeq[TvSerie]]
 
   def getPersonsForTvSerieByTvSerieTitleFuture(tvSeriePrimaryTitle: String): Future[IndexedSeq[Person]]

src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamFileProcessingImpl.scala

 
   implicit val logger: Logger = Logger(LoggerFactory.getLogger(this.getClass))
 
-  override def getPersonByIdFuture(personID: String): Future[Person] = {
+  override def getPersonByIdFuture(personID: String): Future[Option[Person]] = {
     val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = nameBasics)
 
     val start: Long = System.currentTimeMillis()
-    val personFuture: Future[Person] = source
+    val personFuture: Future[Option[Person]] = source
       .via(flow = filterPersonByIdFlow(personID = personID))
-      .runWith(Sink.head[Person])
+      .runWith(Sink.headOption[Person])
 
     personFuture.andThen({
       case Failure(exception) => logger.error(s"!${exception.printStackTrace()}")
     personFuture
   }
 
-  override def getPersonByNameFuture(primaryName: String): Future[Person] = {
+  override def getPersonByNameFuture(primaryName: String): Future[Option[Person]] = {
     val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = nameBasics)
 
     val start: Long = System.currentTimeMillis()
-    val personFuture: Future[Person] = source
+    val personFuture: Future[Option[Person]] = source
       .via(flow = filterPersonByNameFlow(primaryName = primaryName))
-      .runWith(Sink.head[Person])
+      .runWith(Sink.headOption[Person])
 
     personFuture
   }
 
-  override def getTvSerieByIdFuture(tvSerieID: String): Future[TvSerie] = {
+  override def getTvSerieByIdFuture(tvSerieID: String): Future[Option[TvSerie]] = {
     val source: Source[Map[String, String], _] = buildAndValidateSource(inputFile = titleBasics)
 
     val start: Long = System.currentTimeMillis()
-    val tvSerieFuture: Future[TvSerie] = source
+    val tvSerieFuture: Future[Option[TvSerie]] = source
       .via(flow = filterTvSerieByIdFlow(tvSerieID = tvSerieID))
-      .runWith(Sink.head[TvSerie])
+      .runWith(Sink.headOption[TvSerie])
 
     tvSerieFuture.onComplete({
       case Failure(exception) => logger.info(s"$exception")
-      case Success(value: TvSerie) =>
+      case Success(value: Option[TvSerie]) =>
         logger.info(s"$value")
         logger.info(s"SUCCESS, elapsed time:${(System.currentTimeMillis() - start) / 1000} sec")
     })