CompletionStage.thenCompose ne s'exécute pas en série

Question

J'essaie d'utiliser java 8 CompletionStages pour exécuter en série 2 méthodes asynchrones, afin que la seconde ne soit pas exécutée si la première échoue. Mais quand je l'appelle thenCompose, la fonction passée en semble commencer avant que la fonction précédente est terminée (par exemple:. Les deux fonctions exécuter par erreur en parallèle Voici le code:

public CompletionStage<Graph> create(Payload payload) { 
    CompletionStage<BlobInfo> fileFuture = createFile(payload); 
    CompletionStage<Entity> metadataFuture = createMetadata(payload); 
    return fileFuture 
     .thenCompose(ignore -> metadataFuture) 
     .thenApply(entity -> 
      buildFromEntity(objectMapper, entity)); 
    } 

    public CompletionStage<BlobInfo> createFile(Payload payload) { 
    return CompletableFuture.supplyAsync(() -> { 
     try { 
     return 
      storage.create(
       BlobInfo 
        .newBuilder(payload.bucket, payload.name) 
        .build(), 
       payload.data.getBytes()); 
     } catch (StorageException e) { 
     LOG.error("Failed to write to storage: " + e); 
     throw new RequestHandlerException(StatusCode.SERVER_ERROR, 
      "Failed to write to storage."); 
     } 
    }); 
    } 


    public CompletionStage<Entity> createMetadata(Payload payload) { 
    return CompletableFuture.supplyAsync(() -> createSync(payload)); 
    } 

    private Entity createMetadataSync(Payload payload) { 
    Key key = keyFactory.newKey(payload.id); 
    Entity.Builder entityBuilder = GraphPayload.buildEntityFromGraph(payload, key); 
    Entity entity = entityBuilder.build(); 
    LOG.error("Metadata.createSync"); 

    try { 
     datastore.add(entity); 
    } catch (DatastoreException e) { 
     LOG.error("Failed to write initial metadata: " + e); 
     throw new RequestHandlerException(StatusCode.SERVER_ERROR, 
      "Failed to write initial metadata."); 
    } 
    return entity; 
    } 

SORTIE:

16:57:47.530 [ForkJoinPool.commonPool-worker-3] ERROR com.spotify.nfgraphstore.store.FileStore - CreateFile 
16:57:47.530 [ForkJoinPool.commonPool-worker-2] ERROR com.spotify.nfgraphstore.store.MetadataStore - Metadata.createSync 
16:57:47.530 [ForkJoinPool.commonPool-worker-3] ERROR com.spotify.nfgraphstore.store.FileStore - Failed to write initial graph to storage: com.google.cloud.storage.StorageException: X 

La sortie journalisée démontre que Metadata.createSync est exécutée avant que l'exception de stockage ne soit levée.Cette conclusion est également le résultat d'un test (non montré) censé afficher des interactions nulles avec la métadonnée DB si l'écriture dans le fichier la base de données de stockage échoue, ce test échoue parfois, ce qui suggère une condition de concurrence

Donc, je pense que thenCompose ne garantit pas l'exécution en série. Mais tout ce que j'ai lu dans les docs de java suggère que l'exécution devrait être en série: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html#thenCompose-java.util.function.Function-

Est-ce que quelqu'un sait pourquoi l'exécution n'est pas garantie pour être sérielle, ou recommande d'autres fonctions qui pourraient fonctionner plus comme je l'ai prévu?

Answer 1

L'appel à createMetadata lance la tâche immédiatement, car il n'est pas appelé dans le cadre de l'expression lambda transmise à thenCompose.

Peut-être que vous vouliez faire:

.thenCompose(ignore -> createMetadata(payload))