title: Add custom logic/callback/handler when EOF for JsonFraming in akka-streams
body: <p>I have a flow where I consume the paths to the files in small batches from Kafka pah topics, read the files themselves(big JSON arrays) and write them back to Kafka data topics.</p>
  <p>It looks like this:</p>
  <pre><code>      val fileFlow = Flow[Path].flatMapConcat(HdfsSource.data(fs, _))
          .via(JsonFraming.objectScanner(Int.MaxValue))
  
        Consumer
          .committableSource(newConsumerSettings, Subscriptions.topics(inputTopicNames))
          .map(value =&gt; value)
          .grouped(kafkaConsumerBatch)
          .flatMapConcat(paths =&gt; Source(paths))
          .map(path =&gt; new Path(path.record.value().get(&quot;fullPath&quot;).asInstanceOf[String]))
          //Based on: https://github.com/akka/alpakka/blob/v3.0.0/doc-examples/src/test/scala/akka/stream/alpakka/eip/scaladsl/PassThroughExamples.scala#L72-L92
          .via(PassThroughFlow(fileFlow))
          .map { case (bytes, path) =&gt; (bytes, entityConfigMap(getCountryPrefix(path))) }
          .map(bytesAndPath =&gt; (bytesAndPath._1.utf8String.parseJson.asJsObject, bytesAndPath._2))
          .map { case (bytes, entityConfig) =&gt; (toGenericRecord(bytes, entityConfig), entityConfig) }
          .map { case (record, entityConfig) =&gt;
            producerMessagesToTopic.mark()
            ProducerMessage.single(
              new ProducerRecord[NotUsed, GenericRecord](getDataTopicName(entityConfig), record),
              passThrough = entityConfig)
          }
          .via {
            akka.kafka.scaladsl.Producer.flexiFlow(prodSettings)
          }
  ....More logic for logging and running/materializing the flow
  </code></pre>
  <p>Now, the problem is that, as I said, those JSON files are big, so I can't take the entire file content, frame it into separate objects, store them all to Kafka and commit only after that. I mean, that's what I need to do, but I also  need to control the offset commit based on the EOF event.</p>
  <p>I want to let Producer send data over to Kafka in its own pace and regardless of its configs, but somehow inject my custom logic into an EOF event. Maybe something like a passThrough field to signify that the file has been fully consumed and we can now commit offset for the upstream path topic.<br />
  <code>objectScanner</code> has a <code>GraphStageLogic</code> inside its definition that has <code>onUpstreamFinish</code> callback, but there's no direct access to it for overriding. And classes like <code>SimpleLinearGraphStage</code>, <code>JsonObjectParser</code> are marked as Internal APIs.</p>