ReduceFnRunner overhead

More context can be found in discussion here:

[http://mail-archives.apache.org/mod_mbox/beam-dev/201904.mbox/%3CCAOUjMkyKV8npYJfS_PF3Gzo=vwOmB2FRzUtE81ZsrxnM13Tisw@mail.gmail.com%3E](http://mail-archives.apache.org/mod_mbox/beam-dev/201904.mbox/%3CCAOUjMkyKV8npYJfS_PF3Gzo=vwOmB2FRzUtE81ZsrxnM13Tisw@mail.gmail.com%3E)

I have found out on FlinkRunner streaming pipeline there is an overhead associated with processing each element at:
- ReduceFnRunner.scheduleGarbageCollectionTimer() for window
- tracking PaneInfo

This cause quite some trash for JVM GC. At least second option also involves interaction with state backend.

Relevant stacks for illustration:

ScheduleGarbageCollectionTimer
```

...
[ 3] org.apache.flink.streaming.api.operators.HeapInternalTimerService.deleteEventTimeTimer
[
4] org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.deleteTimer
[
5] org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.cancelPendingTimerById
[
6] org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.setTimer
[
7] org.apache.beam.runners.core.ReduceFnContextFactory$TimersImpl.setTimer
[ 8] org.apache.beam.runners.core.ReduceFnRunner.scheduleGarbageCollectionTimer
[
9] org.apache.beam.runners.core.ReduceFnRunner.processElement
[10] org.apache.beam.runners.core.ReduceFnRunner.processElements
[11]
org.apache.beam.runners.core.GroupAlsoByWindowViaWindowSetNewDoFn.processElement
[12] org.apache.beam.runners.core.GroupAlsoByWindowViaWindowSetNewDoFn$DoFnInvoker.invokeProcessElement
[13]
org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement
[14] org.apache.beam.runners.core.SimpleDoFnRunner.processElement
[15]
org.apache.beam.runners.core.LateDataDroppingDoFnRunner.processElement
[16] org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement
[17]
org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.processElement
[18] org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput
[19]
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run
[20] org.apache.flink.streaming.runtime.tasks.StreamTask.invoke
[21]
org.apache.flink.runtime.taskmanager.Task.run
[22] java.lang.Thread.run

```

PaneInfoTracker: Read
```

[ 0] org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.get
[ 1] org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.get
[
2] org.apache.flink.runtime.state.heap.HeapValueState.value
[ 3] org.apache.beam.runners.flink.translation.wrappers.streaming.state.FlinkStateInternals$FlinkValueState.read
[
4] org.apache.beam.runners.core.PaneInfoTracker$1.read
[ 5] org.apache.beam.runners.core.PaneInfoTracker$1.read
[
6] org.apache.beam.runners.core.ReduceFnRunner.onTrigger
[ 7] org.apache.beam.runners.core.ReduceFnRunner.emit
[
8] org.apache.beam.runners.core.ReduceFnRunner.processElements
[ 9] org.apache.beam.runners.core.GroupAlsoByWindowViaWindowSetNewDoFn.processElement
[10]
org.apache.beam.runners.core.GroupAlsoByWindowViaWindowSetNewDoFn$DoFnInvoker.invokeProcessElement
[11]
org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement
[12] org.apache.beam.runners.core.SimpleDoFnRunner.processElement
[13]
org.apache.beam.runners.core.LateDataDroppingDoFnRunner.processElement
[14] org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement
[15]
org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.processElement
[16] org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput
[17]
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run
[18] org.apache.flink.streaming.runtime.tasks.StreamTask.invoke
[19]
org.apache.flink.runtime.taskmanager.Task.run
[20] java.lang.Thread.run

```

PaneInfoTracker: Write
```

[ 0] org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.putEntry
[ 1] org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.put
[
2] org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.put
[ 3] org.apache.flink.runtime.state.heap.HeapValueState.update
[
4] org.apache.beam.runners.flink.translation.wrappers.streaming.state.FlinkStateInternals$FlinkValueState.write
[
5] org.apache.beam.runners.core.PaneInfoTracker.storeCurrentPaneInfo
[ 6] org.apache.beam.runners.core.ReduceFnRunner.lambda$onTrigger$1
[
7] org.apache.beam.runners.core.ReduceFnRunner$$Lambda$101.211931975.output
[ 8] org.apache.beam.runners.core.ReduceFnContextFactory$OnTriggerContextImpl.output
[
9] org.apache.beam.runners.core.SystemReduceFn.onTrigger
[10] org.apache.beam.runners.core.ReduceFnRunner.onTrigger
[11]
org.apache.beam.runners.core.ReduceFnRunner.emit
[12] org.apache.beam.runners.core.ReduceFnRunner.processElements
[13]
org.apache.beam.runners.core.GroupAlsoByWindowViaWindowSetNewDoFn.processElement
[14] org.apache.beam.runners.core.GroupAlsoByWindowViaWindowSetNewDoFn$DoFnInvoker.invokeProcessElement
[15]
org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement
[16] org.apache.beam.runners.core.SimpleDoFnRunner.processElement
[17]
org.apache.beam.runners.core.LateDataDroppingDoFnRunner.processElement
[18] org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement
[19]
org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.processElement
[20] org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput
[21]
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run
[22] org.apache.flink.streaming.runtime.tasks.StreamTask.invoke
[23]
org.apache.flink.runtime.taskmanager.Task.run
[24] java.lang.Thread.run

```

Imported from Jira [BEAM-7204](https://issues.apache.org/jira/browse/BEAM-7204). Original Jira may contain additional context.
Reported by: JozoVilcek.