StreamConverters.asInputStream not workable when using broadcast with ByteString source?

[mdedetrich]

Currently I have this task where I have a Source (which is a ByteString) and which I want to fan out via a Broadcast to the other flows. One flow goes directly into a Sink (which I will call an upload sink) and the other flow needs to be needs to convert the ByteString Source to an Java InputStream since its going to an external API that only accepts InputStream, specifically this is Apache tika which has a detect function as follows

    /**
     * Detects the media type of the given document. The type detection is
     * based on the content of the given document stream.
     * <p>
     * If the document stream supports the
     * {@link InputStream#markSupported() mark feature}, then the stream is
     * marked and reset to the original position before this method returns.
     * Only a limited number of bytes are read from the stream.
     * <p>
     * The given document stream is <em>not</em> closed by this method.
     *
     * @param stream the document stream
     * @return detected media type
     * @throws IOException if the stream can not be read
     */
    public String detect(InputStream stream) throws IOException {
        return detect(stream, new Metadata());
    }

(note the fact that this function does NOT close the InputStream, you have to do this yourself)

Pekko Streams provides a StreamConverters.asInputStream() function which is a Sink[ByteString, InputStream]. As discussed at akka/akka-core#23187 (comment) and noted in the docs, you need to be careful when using the materialized InputStream as its a blocking interface, which means that if you do something like this

StreamConverters.asInputStream().mapMaterializedValue { inputStream ->
  Tika().detect(inputStream)
}

It will create a deadlock as materialized values need to return immediately and the Tika().detect(inputStream) call will block. Due to this its advised that you materialize the StreamConverters.asInputStream() stream immediately to get the InputStream and then use the InputStream as desired.

My current predicament is that while this works if you have a single trivial flow, when you have a broadcast style fanout it causes issues one way or another. For example one way to solve this issue would be as follows

/**
 * Uploads a file while at the same time computing the media type
 */
fun <Mat> computeMediaTypeAndUpload(
    source: Source<ByteString, *>,
    uploadSink: Sink<ByteString, CompletionStage<Mat>>
): Source<Pair<Optional<String>, Mat>, CompletionStage<NotUsed>> {
    return Source.fromMaterializer { mat, attr ->
        /**
         * Due to InputStream being blocking, we CANNOT immediately read from
         * the InputStream as it will cause a deadlock/timeout error, see
         * https://github.com/akka/akka/issues/23187#issuecomment-317646764
         *
         * To get around this, we run the sink immediately to get the InputStream
         * and then we process it at a later stage.
         */
        val inputStreamWithMat = source.alsoToMat(StreamConverters.asInputStream(), Keep.right())
            .toMat(uploadSink, Keep.both()).withAttributes(attr).run(mat)

        val inputStream = inputStreamWithMat.first()

        val blockingIODispatcher =
            mat.system().dispatchers().lookup(Dispatchers.DefaultBlockingDispatcherId())

        Source.lazyCompletionStage {
            val computedFileMediaTypeFuture = CompletableFuture.supplyAsync({
                try {
                    Optional.ofNullable(Tika().detect(inputStream))
                } catch (ex: IOException) {
                    logger.error("Error detecting media type", ex)
                    Optional.empty()
                }
            }, blockingIODispatcher)

            val matFuture = inputStreamWithMat.second().toCompletableFuture()

            CompletableFuture.allOf(
                computedFileMediaTypeFuture, matFuture
            ).thenApply {
                Pair(computedFileMediaTypeFuture.join(), matFuture.join())
            }
        }
    }
}

The premise here is to use a Java CompletableFuture to wrap the Tika().detect(inputStream) computation so that .mapMaterializedValue returns immediately. Doing this however still deadlocks, to avoid this one can change

try {
    Optional.ofNullable(Tika().detect(inputStream))
} catch (ex: IOException) {
    logger.error("Error detecting media type", ex)
    Optional.empty()
}

to

try {
    Optional.ofNullable(Tika().detect(inputStream))
} catch (ex: IOException) {
    logger.error("Error detecting media type", ex)
    Optional.empty()
} finally {
    inputStream.close()
}

The inputStream.close() appears to force demand on the stream, fixing the deadlock however doing this causes other issues, namely that this exception ends up being thrown.

java.util.concurrent.CompletionException: org.apache.pekko.stream.SubscriptionWithCancelException$NoMoreElementsNeeded$                                                                                                 at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
at java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source)
at java.base/java.util.concurrent.CompletableFuture$BiRelay.tryFire(Unknown Source)
at java.base/java.util.concurrent.CompletableFuture$CoCompletion.tryFire(Unknown Source)
at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(Unknown Source)
at scala.concurrent.impl.FutureConvertersImpl$CF.apply(FutureConvertersImpl.scala:26)
at scala.concurrent.impl.FutureConvertersImpl$CF.apply(FutureConvertersImpl.scala:23)
at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:484)
at scala.concurrent.ExecutionContext$parasitic$.execute(ExecutionContext.scala:222)
at scala.concurrent.impl.Promise$Transformation.submitWithValue(Promise.scala:429)

With this error the logic of the stream still works, but it appears that Pekko Streams cannot properly handle eagerly closing the InputStream (even though calling inputStream.close() multiple times is perfectly valid). More importantly, doing inputStream.close() like this is a workaround/hack because as described in the docs

  /**
   * Creates a Sink which when materialized will return an [[java.io.InputStream]] which it is possible
   * to read the values produced by the stream this Sink is attached to.
   *
   * This method uses a default read timeout, use [[#inputStream(FiniteDuration)]] or [[#inputStream(java.time.Duration)]] to explicitly
   * configure the timeout.
   *
   * This Sink is intended for inter-operation with legacy APIs since it is inherently blocking.
   *
   * You can configure the internal buffer size by using [[pekko.stream.ActorAttributes]].
   *
   * The [[InputStream]] will be closed when the stream flowing into this [[Sink]] completes, and
   * closing the [[InputStream]] will cancel this [[Sink]].
   */
  def asInputStream(): Sink[ByteString, InputStream] = new Sink(scaladsl.StreamConverters.asInputStream())

Pekko Streams is meant to handle the resource cleanup of the underlying stream and so you shouldn't have to call .close() like this.

The thing is I have tried various different implementations of this (i.e. using preMaterialize, custom graph with GraphDSL + Broadcast, Flow.fromSinkAndSource, fromMaterializer, Source.lazyCompletionStage vs Source.completionStage etc etc) and nothing seems to have solve the underlying issue. Either you have a deadlock (if you don't do inputStream.close()) or if you do have inputStream.close() then you get the SubscriptionWithCancelException$NoMoreElementsNeeded exception (note that I haven't been able to reproduce this exception with a trivial example locally but it occurs in production when computeMediaTypeAndUpload is composed as part of a bigger stream).

Of course one way to "solve" the problem is to just do

source.alsoToMat(uploadSink, Keep.right())
  .toMat(Sink.fold(ByteString.emptyByteString()) { a, b -> a.concat(b) }, Keep.both())
  .run(system)

to avoid using InputStream entirely but this basically kills the streaming part of the flow into Apache Tika, as you just evaluate the entire stream into a single large String where as my aim is to make sure that both flows in BroadCast are streaming based.

EDIT: I stated earlier that with the SubscriptionWithCancelException$NoMoreElementsNeeded the logic still worked, this isn't actually the case.

[ptrdom]

StreamConverters.asInputStream does not seem like a right tool for your problem, at least to me. Higher-level stream operators are not always good enough, I would likely consider using a custom graph stage to implement stateful processing that takes the ByteString elements as input and returns the result of Tika().detect as output.

[mdedetrich]

Looking at the InputStreamStage, it does look like it may have been written with InputStream semantics - supporting a one time read and that's it. For Fan-Out scenarios, it feels better to materialize a ByteString which is re-readable - you can get more than 1 InputStream from the ByteString and read the data more than once.

Indeed but if I understand you correctly, thats why there is Broadcast (which alsoToMat uses internally), the entire point of that is to fan out the source byte stream so that one copy is sent to the uploadSink and another copy is sent to the StreamConverters.inputStream()'s inputStream so that this shouldn't be an issue since InputStream will be working on a copy of the original source.

Or maybe I am misunderstanding something here?

EDIT: I just did some searching and maybe I should be using BroadcastHub.Sink instead???

[mdedetrich]

So I did some more testing and I can confirm that when using inputStream().close within the finally block, the random SubscriptionWithCancelException$NoMoreElementsNeeded does actually effect the business logic as the stream gets terminated before it can complete properly

[mdedetrich]

@pjfanning Unfortunately its not really possible/easy to test with Pekko 1.0.0 because we rely on new features and this issue only happens in production.

My current suspicion for the underlying issue is that there is a race condition between inputStream.close() and Tika().detect(inputStream) where for some reason the inputStream.close() happens before Tika().detect(inputStream).

I don't know how/why this happens but it does explain the current behaviour.

[pjfanning]

Would it be possible to write a custom stage that materialises a ByteString and then you can call asInputStream on the ByteString instance? This InputStream is an independent InputStream and is usable until it is closed in its own right.

[mdedetrich]

Would it be possible to write a custom stage that materialises a ByteString and then you can call asInputStream on the ByteString instance?

I am not sure what you mean by "materialize a ByteString" because the source streams don't have ByteString in the materialization, instead its part of the actual stream.

As an example, FileIO.fromPath which creates a source from a file has the following type signature Source[ByteString, Future[IOResult]]. In this case the materialized value is Future[IOResult] and not ByteString as the ByteString is being streamed behind the scenes and the Future[IOResult] is the materialized value which represents whether the file operation is successful (or not) and the reason why its in a Future is because materialized values have to be returned before the stream starts and so in this specific case the Future would get completed when the stream finishes (successfully or not).

However if by "materialize a ByteString" you mean evaluating the Source[ByteString, _] to get the bytestring out, yes you can do that but then you are essentially evaluated the whole stream so its not really a stream anymore.

I guess you could jerry rig a Sink.foreach so that you could manually push ByteString chunks into a manually created InputStream but this would just be a more primitive implementation of what currently exists.

I have another theory which is that .alsoToMat which internally uses Broadcast (and this is what should be copying the ByteString in the fanout) has eagerCancel set to true and this is what may be causing issues. I will try this out next week as I can't do it right now.