diff options
| author | S7evinK <tfaelligen@gmail.com> | 2020-10-27 15:11:37 +0100 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2020-10-27 14:11:37 +0000 |
| commit | d5675feb96b7e3996ca2768fddb9f534dc268816 (patch) | |
| tree | f2b458c6bbe9a89d3e28474ffefa38614cdb7bfa /roomserver | |
| parent | c5888bb64cc4b393bccf0fa79d667680144902d0 (diff) | |
Add possibilty to configure MaxMessageBytes for sarama (#1563)
* Add configuration for max_message_bytes for sarama
* Log all errors when sending multiple messages
Signed-off-by: Till Faelligen <tfaelligen@gmail.com>
* Add missing config
* - Better comments on what MaxMessageBytes is used for
- Also sets the size the consumer may use
Diffstat (limited to 'roomserver')
| -rw-r--r-- | roomserver/internal/input/input.go | 8 |
1 files changed, 7 insertions, 1 deletions
diff --git a/roomserver/internal/input/input.go b/roomserver/internal/input/input.go index d340ac21..99c15f77 100644 --- a/roomserver/internal/input/input.go +++ b/roomserver/internal/input/input.go @@ -102,7 +102,13 @@ func (r *Inputer) WriteOutputEvents(roomID string, updates []api.OutputEvent) er Value: sarama.ByteEncoder(value), } } - return r.Producer.SendMessages(messages) + errs := r.Producer.SendMessages(messages) + if errs != nil { + for _, err := range errs.(sarama.ProducerErrors) { + log.WithError(err).WithField("message_bytes", err.Msg.Value.Length()).Error("Write to kafka failed") + } + } + return errs } // InputRoomEvents implements api.RoomserverInternalAPI |