then:
        collectFiles(filteredOutputs) == [outputDir]

        when:
        def outputDirFile = outputDir.file("in-output-dir").createFile()
        virtualFileSystem.invalidateAll()
        def afterExecution = snapshotOutput(outputDir)
        filteredOutputs = filterOutputAfterExecution(EMPTY, beforeExecution, afterExecution)
        then:
        collectFiles(filteredOutputs) == [outputDir, outputDirFile]

                                                   TF_Status* status);

// Aborts all ongoing collectives with the specified status. After abortion,
// subsequent collectives will error with this status immediately. To reset the
// collectives, create a new EagerContext.
//
// This is intended to be used when a peer failure is detected.

  EXPECT_NE(TF_GetCode(status.get()), TF_CANCELLED);
  EXPECT_EQ(TF_GetCode(status.get()), TF_INVALID_ARGUMENT);
  EXPECT_THAT(TF_Message(status.get()), HasSubstr("assertion failed"));

  // Note that future collectives with the same context do not work at the
  // moment; once canceled, the collective executor requires the program to be
  // restarted / context to be reset.
}

TEST(PARALLEL_DEVICE_LIB, TestDifferentShapes) {

                        const int in_flight_nodes_limit)
      : status_(TF_NewStatus()),
        // If the context's default exector is set to async, re-using that in
        // each thread would cause collectives to deadlock. For consistency we
        // create a new sync executor for every thread.
        //
        // TODO(allenl): We should have an async API that works with the
        // parallel device.

  }
  return &it->second;
}

void RunInThreadPoolIfCollectivesPresent(
    const XlaCompiler::CompilationResult& compilation_result,
    std::function<void()> execution_fn) {
  // If we are using collectives, we need to run in a separate threadpool.
  if (compilation_result.collective_info.has_value()) {
    GetOrCreateThreadPoolForCollective(*compilation_result.collective_info)
        ->Schedule(execution_fn);
  } else {