inloco/kafka-elasticsearch-injector

Golang app to read records from a set of kafka topics and write them to an elasticsearch cluster

kafka-elasticsearch-injector's Issues

Nil Map when unmarshalling Json

On function JsonMessageToRecord, the msg.Value may be a JSON literal null. In that case, according to Go's documentation:

Unmarshal first handles the case of the JSON being the JSON literal null. In that case, Unmarshal sets the pointer to nil

Given that, jsonValue will be a nil map and you cannot write any value to nil map.

It would be better to handle any errors before writing any values to the map and also to check if the map is not nil, as JSON literal null is a valid input.

Remove deprecated https://github.com/bsm/sarama-cluster

https://github.com/bsm/sarama-cluster is now deprecated. We should replace it with https://github.com/Shopify/sarama

Synchronous flush

KAFKA_CONSUMER_BATCH_SIZE controls when the in-memory events are flushed to ES.

However some topics may have small throughput, not having their events flushed at all.

It would be nice to have a second parameter controlling a recurrent tick so the in-memory events are flushed even when they do not meet the batch size requirement

SchemaRegistry cache my fail in concurrent miss scenario

I this following code may not work in concurrent miss scenario
https://github.com/inloco/kafka-elasticsearch-injector/blob/master/src/kafka/encode_decode.go#L115-L142

the solution described here with ChanLocker pattern might fix it.
https://lakefs.io/2020/09/23/in-process-caching-in-go-scaling-lakefs-to-100k-requests-second/