Kafka::MessageSizeTooLarge

config:
```

@type kafka2

BROKERS

username
password
scram_mechanism sha256
sasl_over_ssl true

ssl_ca_cert [CERT]

@type json

@type file
path /mnt/pos/buffers
chunk_limit_size 64MB
total_limit_size 1024MB
flush_mode interval
flush_interval 3s
flush_thread_count 4
retry_type exponential_backoff
retry_max_interval 30
retry_forever

max_send_limit_bytes 900000

default_topic messages
compression_codec snappy

max_send_retries 4
required_acks -1

exclude_topic_key true
exclude_partition true
exclude_partition_key true
exclude_message_key true

```

Fluentd works fine for a while, then gets stuck on a large chunk. Usually seems to happen when one of the logs it's matching has a spike in throughput

```
2020-02-27 21:13:23 +0000 [warn]: #0 Send exception occurred: Kafka::MessageSizeTooLarge │
│ 2020-02-27 21:13:23 +0000 [warn]: #0 Exception Backtrace : /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/protocol.rb:160:in `handle_error' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/produce_operation.rb:153:in `block in handle_response' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/protocol/produce_response.rb:36:in `block (2 levels) in each_partition' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/protocol/produce_response.rb:35:in `each' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/protocol/produce_response.rb:35:in `block in each_partition' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/protocol/produce_response.rb:34:in `each' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/protocol/produce_response.rb:34:in `each_partition' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/produce_operation.rb:144:in `handle_response' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/produce_operation.rb:133:in `block in send_buffered_messages' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/produce_operation.rb:105:in `each' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/produce_operation.rb:105:in `send_buffered_messages' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/produce_operation.rb:62:in `block in execute' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/instrumenter.rb:23:in `instrument' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.7.10/lib/kafka/produce_operation.rb:53:in `execute' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-kafka-0.12.3/lib/fluent/plugin/kafka_producer_ext.rb:210:in `block in deliver_messages_with_retries' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-kafka-0.12.3/lib/fluent/plugin/kafka_producer_ext.rb:200:in `loop' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-kafka-0.12.3/lib/fluent/plugin/kafka_producer_ext.rb:200:in `deliver_messages_with_retries' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-kafka-0.12.3/lib/fluent/plugin/kafka_producer_ext.rb:126:in `deliver_messages' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluent-plugin-kafka-0.12.3/lib/fluent/plugin/out_kafka2.rb:265:in `write' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.9.2/lib/fluent/plugin/output.rb:1133:in `try_flush' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.9.2/lib/fluent/plugin/output.rb:1439:in `flush_thread_run' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.9.2/lib/fluent/plugin/output.rb:461:in `block (2 levels) in start' │
│ /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.9.2/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create' │
│ 2020-02-27 21:13:23 +0000 [info]: #0 initialized kafka producer: fluentd │
│ 2020-02-27 21:13:23 +0000 [warn]: #0 failed to flush the buffer. retry_time=0 next_retry_seconds=2020-02-27 21:13:24 +0000 chunk="59f92c5b59f6035ea8236814fb465867" error_class=Kafka::MessageSizeTooLarge error="Kafka::MessageSizeTooLarge" │
│ 2020-02-27 21:13:23 +0000 [warn]: #0 suppressed same stacktrace │
│ 2020-02-27 21:13:24 +0000 [warn]: #0 retry succeeded. chunk_id="59f9530bf1269c3daa29759ad3f94b15"
```

The max message bytes limit on the kafka brokers is 2000000.