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

[Claudenw]

Perhaps we could write the records to local files and then upload to permanent storage on a flush. In this way we should avoid the issue where we crash before Kafka has not committed the record but we have written it to storage. In this case I think a restart will result in duplicate records in the storage.

[mkeskells]

That doesn't help with back pressure. The whole idea is to buffer less

Storing files locally makes the flush a bit quicker, but writing them asynchronous is a better solution

The back pressure is to induce the flush earlier, and limit the rate of consumption

Depending on the configuration, the file may or may not be available after a restart. It could be ephemeral storage, so buffering that is just memory, that needs another threshold to monitor and cause back pressure from

So writing to file doesn't solve the issue and provides a different resource to be managed and cause back pressure