Introduce back pressure on record writing

[mkeskells]

I would like to introduce a feature that ensure that we don't buffer too many records in memory before we flush to files.

Theer is a basic implementation of this in #319 but this is intermingled and depends on several other features in the same PR

I think that we need to have some support for back pressure, and this can evolve into a more sophisticated solution

The Goals are

• reduce unbounded memory pressure
• eliminate OOMs due to buffered records not yet written

Its isn't expected that this can be done without some configuration changes

related to
#270
#259

[Claudenw]

@mkeskells can you document your back pressure ideas here?

[mkeskells]

The concept of back pressure is to limit the work in progress

Effectively - ensure that we don't consume too much data via messages put until we have removed some from the heap, by writing to external file, and removing the records from reachability

The target solution, which I will document more in another discussion, involves rewriting the was that records are separated into files and the files are written ( as you can see in #319) but this has a lot of other noise in it as well

In the short term it seems simple that we request a flush if the number of records that have been buffered exceeds a threshold

So on a call to put we keep track of the number of records
If the total records buffered is greater than some configured value, we request flush
via context.requestCommit()' where context is the protected SinkTaskContext context; inSinkTask`

This is turn causes the iteration loop in WorkerSinkTask to commitOffsets(), which call preCommit() and currently that calls flush()

on flush we write the records as normal and reset the counter

The above is simple, and should stop OOM, unless we get too many records in a single put(), but we can limit the batch size via out standard kafka properties

The above is not a good long term solution IMHO, but it should kill the OOM issue. for the code straw man see #333 as a PR for discussion