Add possibilty to configure MaxMessageBytes for sarama
* Add configuration for max_message_bytes for sarama
* Log all errors when sending multiple messages
* Add missing config
* - Better comments on what MaxMessageBytes is used for
Also sets the size the consumer may use
MODIFY
dendrite-config.yaml
dendrite-config.yaml
@@ -76,6 +76,12 @@ global:
# Kafka.
use_naffka: true
+ # The max size a Kafka message is allowed to use.
+ # You only need to change this value, if you encounter issues with too large messages.
+ # Must be less than/equal to "max.message.bytes" configured in Kafka.
+ # Defaults to 8388608 bytes.
+ # max_message_bytes: 8388608
+
# Naffka database options. Not required when using Kafka.
naffka_database:
connection_string: file:naffka.db

MODIFY
internal/config/config_kafka.go
internal/config/config_kafka.go
@@ -24,6 +24,9 @@ type Kafka struct {
UseNaffka bool `yaml:"use_naffka"`
// The Naffka database is used internally by the naffka library, if used.
Database DatabaseOptions `yaml:"naffka_database"`
+ // The max size a Kafka message passed between consumer/producer can have
+ // Equals roughly max.message.bytes / fetch.message.max.bytes in Kafka
+ MaxMessageBytes *int `yaml:"max_message_bytes"`
}
func (k *Kafka) TopicFor(name string) string {
@@ -36,6 +39,9 @@ func (c *Kafka) Defaults() {
c.Addresses = []string{"localhost:2181"}
c.Database.ConnectionString = DataSource("file:naffka.db")
c.TopicPrefix = "Dendrite"
+
+ maxBytes := 1024 * 1024 * 8 // about 8MB
+ c.MaxMessageBytes = &maxBytes
}
func (c *Kafka) Verify(configErrs *ConfigErrors, isMonolith bool) {
@@ -50,4 +56,5 @@ func (c *Kafka) Verify(configErrs *ConfigErrors, isMonolith bool) {
checkNotZero(configErrs, "global.kafka.addresses", int64(len(c.Addresses)))
}
checkNotEmpty(configErrs, "global.kafka.topic_prefix", string(c.TopicPrefix))
+ checkPositive(configErrs, "global.kafka.max_message_bytes", int64(*c.MaxMessageBytes))
}

MODIFY
internal/setup/kafka/kafka.go
internal/setup/kafka/kafka.go
@@ -17,12 +17,17 @@ func SetupConsumerProducer(cfg *config.Kafka) (sarama.Consumer, sarama.SyncProdu
// setupKafka creates kafka consumer/producer pair from the config.
func setupKafka(cfg *config.Kafka) (sarama.Consumer, sarama.SyncProducer) {
- consumer, err := sarama.NewConsumer(cfg.Addresses, nil)
+ sCfg := sarama.NewConfig()
+ sCfg.Producer.MaxMessageBytes = *cfg.MaxMessageBytes
+ sCfg.Producer.Return.Successes = true
+ sCfg.Consumer.Fetch.Default = int32(*cfg.MaxMessageBytes)
+
+ consumer, err := sarama.NewConsumer(cfg.Addresses, sCfg)
if err != nil {
logrus.WithError(err).Panic("failed to start kafka consumer")
}
- producer, err := sarama.NewSyncProducer(cfg.Addresses, nil)
+ producer, err := sarama.NewSyncProducer(cfg.Addresses, sCfg)
if err != nil {
logrus.WithError(err).Panic("failed to setup kafka producers")
}

MODIFY
roomserver/internal/input/input.go
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

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