SinkGroupSpec, DeploymentUnit, MaxBytesPerBatch

[alok87]

Brings up new specification to define sinkGroup based batcher and loader configuration. Introduces the concept of deployment unit which the user can specify to solve the scaling issues described in #167.maxBytesPerBatch replaces maxSize as it makes all the topics operating in the same pod take the same amount of resources, paving the way for scaling being possible. As scaling requires homegeneous resource consumption across multiple topics.

• SinkGroup spec and Deployment unit in the API. Batcher: Processing speed optimisation and scalability #167 Separate resource requirement for sinkgroups #138 Batcher first time memory problem #136
• Batcher changes for the new spec with backward compatibility.
• Loader changes for the new spec with backward compatibility.
• Use resource.Quantity in the api spec and define it for MaxBytesPerBatch as MaSizePerBatch so that values can be provided by the user in Mi, Ki, Gi, bytes..
• MaxBytesPerBatch is better than maxSize. MaxSize gets deprecated in favour of MaxBytesPerBatch. This config makes all fat and lean tables behave the same way in the Redshiftbatcher. They both take the same amount of memory and the scaling becomes easier now. Related Batcher first time memory problem #136 Batcher: Processing speed optimisation and scalability #167 Also added for loader as a sum of all batch size.
• Added prometheus metrics for bytes processed for both batcher and loader.
• Refactor out realtime calculation
• Fixes Kafka redshiftbatcher s3 path is shared among diff rsk objects #151
• Solve the allocation problem as described here SinkGroupSpec, DeploymentUnit, MaxBytesPerBatch #172 (comment)
• Set defaults by sinkgroup (later update defaults after some days)
• Set resources of realtime topics as Main
• End to end tests

[alok87]

Live with backward compatibility

[alok87]

Old spec is compatible, tested OK.

[alok87]

Metric is a problem, values being Gauge and not being set in some conditions never go down to 0

[alok87]

Two problems have come out after testing with 300 tables:

1. The distribution of units is not same. Big topics are landing into the same unit, making some unit slow. The distribution of topics to units should be uniform.
2. The batcher units keep on waiting and running even when the lag has come down, and it is waiting for the loader lag to come down.

[alok87]

Things are not auto recovering from

E0325 12:03:27.620298       1 batch_processor.go:498] ts.fabric.vn_call_forwardings, error(s) occured in processing (sending err)
I0325 12:03:27.620305       1 batch_processor.go:176] ts.fabric.vn_call_forwardings: batch processing gracefully shutingdown

[alok87]

Releases are getting stuck.

[alok87]

cpu_pprof.pdf

CPU Profiling

Update: opened a separate issue to solve this #173

[alok87]

if maxReloadingTopics is reduced it does not take effect.

[alok87]

sometimes batcher is stuck in session closure loop, mostly related to MaxProcessingTime, need to be checked
looks dupe of #172 (comment)

Error while closing connection to broker broken pipe

Update: trying internal kafka listners (Strimzi) strimzi/strimzi-kafka-operator#4688

This was fixed after using internal routing.

[alok87]

Loader optimizations

Time

• maxSizePerBatch: v low value causes slowness due to repeated merge
• maxWaitSeconds: if batchSize mbs is small due to maxWait hit due to First schema creation un-necessarily retrys #129 then v small batches are loaded causing slowness
  Update: maxWait should reset after processing is done, so that next batch is not small

Dividing loader into multiple pods

Grouping loader into group of pods based on the lag it has. So that there is minimum shuffling.
(not required, small batches operate at same speed)