feature: callback impl #80
Conversation
|
thank you so much for your contribution @enesyalinkaya |
kafka/producer/producer_batch.go
Outdated
| @@ -136,3 +150,28 @@ func isFatalError(err error) bool { | |||
| } | |||
| return true | |||
| } | |||
|
|
|||
| func (b *Batch) handleWriteError(writeErrors kafka.WriteErrors) { | |||
There was a problem hiding this comment.
can we write unit tests for the below functions?
kafka/producer/producer_batch.go
Outdated
| } | ||
| } | ||
|
|
||
| func convertKafkaMessage(src kafka.Message) message.KafkaMessage { |
There was a problem hiding this comment.
we can move this function to kafka.Message I guess and rename toKafkaMessage
There was a problem hiding this comment.
we cannot move it into kafka.Message there, it is a type belonging to the segmentio library.
callback.go
Outdated
|
|
||
| import "github.com/Trendyol/go-dcp-kafka/kafka/message" | ||
|
|
||
| type callback struct { |
There was a problem hiding this comment.
do we need this struct? maybe we can move the Callback(SinkResponseHandler) interface here and remove below logics?
For Kafka events in our outbox bucket, we want to use go-dcp-kafka to send them. When event is published, we want to remove the event document from the outbox bucket. #77
For this reason, I made a development to add a callback function after the event is published