[jira] [Created] (FLINK-6413) Add stream operator callback to notify about consumed network buffer

["Aljoscha Krettek (JIRA)" <ji...@apache.org>]

Aljoscha Krettek created FLINK-6413:
---------------------------------------

             Summary: Add stream operator callback to notify about consumed network buffer 
                 Key: FLINK-6413
                 URL: https://issues.apache.org/jira/browse/FLINK-6413
             Project: Flink
          Issue Type: Improvement
          Components: DataStream API
            Reporter: Aljoscha Krettek


This is originally motivated by BEAM-1612. Beam has the notion of bundles and allows users to do work at the start/end of each bundle. This could be used for setting up some expensive connection or for batching accesses to some external system. There is also internal optimisation potential because accesses/updates to state could be kept in-memory per bundle/buffer and only afterwards be written to fault-tolerant state.

The bundling induced by the Flink network stack (which depends on the network buffer size and the buffer timeout) seems like a natural fit for this. I propose to add an _experimental_ interface {{BufferConsumedListener}} (or some such name):

{code}
interface BufferConsumedListener {
  void notifyBufferConsumed():
}
{code}

that is invoked in the input processor whenever a network buffer is exhausted: https://github.com/apache/flink/blob/922352ac35f3753334e834632e3e361fbd36336e/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java#L178-L178

The change is very simple, three lines of code would be added:

{code}
if (result.isBufferConsumed()) {
  currentRecordDeserializer.getCurrentBuffer().recycle();
  currentRecordDeserializer = null;
  if (streamOperator instanceof BufferConsumedListener) {
    ((BufferConsumedListener) streamOperator).notifyBufferConsumed():
  }
}
{code}





--
This message was sent by Atlassian JIRA
(v6.3.15#6346)