instructure/canvas-lms · warning

Kafka message purged: #

Error message

Kafka message purged: #{event[:error]&.message}

What it means

This is a warning emitted by the waterdrop Kafka producer monitoring subscriber for the 'message.purged' event. It fires when a buffered message is dropped because it hit 'message.timeout.ms' without receiving a broker acknowledgment; the message is silently discarded. Canvas logs it and increments the kafka_events.messages_purged Statsd counter, and the code comment explicitly says it is worth alerting on.

Solutions

  1. Check Kafka broker health and connectivity from the app host (kafka-topics --describe, broker logs) to find why acks were not delivered.
  2. Raise message.timeout.ms (and delivery-related timeouts) in the waterdrop producer config so normal latency does not trigger purging.
  3. Verify topic replication/ISR is healthy; acks=all with degraded ISR can exceed the timeout.
  4. Add an alert on kafka_events.messages_purged as the code comment suggests, and replay the lost events from the source if data loss matters.

Example fix

# before
config[:message_timeout_ms] = 5000

# after
config[:message_timeout_ms] = 30000
Defensive patterns

Strategy: retry

Validate before calling

if wd.config[:message_timeout_ms] < 30_000
  Rails.logger.warn("kafka message.timeout.ms very low; messages may be purged")
end
# also verify broker reachability before producing
# Kafka::Admin or TCP probe to bootstrap servers

Type guard

def broker_reachable?(seed_servers)
  seed_servers.all? { |host, port| TCPSocket.new(host, port).close; true }
rescue Errno::ECONNREFUSED, SocketError
  false
end

Try / catch

begin
  producer.produce_async(payload)
rescue WaterDrop::Errors::ProduceError => e
  InstStatsd::Statsd.increment("kafka_events.produce_failed")
  # buffer/replay payload later; message.purged means data loss, not an exception
end

Prevention

When it happens

Trigger: Producing a Canvas event via the waterdrop producer whose buffered message exceeds message.timeout.ms waiting for a broker ack, e.g. Kafka brokers unreachable, slow disk flush on the broker, or producer-side retries exhausted before ack timeout.

Common situations: Kafka cluster outage or broker restart during event production; misconfigured message.timeout.ms too low relative to broker latency; network partition between app and Kafka; broker acks settings (e.g. acks=all) with an under-replicated slow topic.

Understand the failure class

Background: Request timed out: what client-side request timeouts mean across libraries (Request timed out, TIMED_OUT, APITimeoutError) — this error's family across 39 libraries.

Related errors

AI-assisted analysis of instructure/canvas-lms@1c9f0bb801 (2026-09-15). Data as JSON: /api/errors/0a41ea8fd71e3ddc. Report an issue: GitHub.

Appendix: source

Thrown at lib/canvas/kafka_events/producer.rb:61

      Rails.logger.warn("Kafka producer close error: #{e.message}")
    end

    private

    def build_waterdrop
      wd = WaterDrop::Producer.new do |c|
        c.kafka = @config.kafka_options
        c.logger = Rails.logger
      end
      wd.monitor.subscribe("error.occurred") do |event|
        InstStatsd::Statsd.distributed_increment("kafka_events.delivery_errors")
        Rails.logger.warn("Kafka delivery failed: #{event[:error]&.message}")
      end
      wd.monitor.subscribe("message.purged") do |event|
        # Fires when a buffered message hits message.timeout.ms without broker ack —
        # the event is silently dropped on the floor. Worth an alert.
        InstStatsd::Statsd.distributed_increment("kafka_events.messages_purged")
        Rails.logger.warn("Kafka message purged: #{event[:error]&.message}")
      end
      wd
    end

    def log_ready
      resolved = Events.topic_keys.map { |key| "#{key}=#{@config.topic_for(key)}" }.join(" ")
      Rails.logger.info("Kafka events producer ready: brokers=#{@config.brokers} #{resolved}")
    end
  end
end

View on GitHub (pinned to 1c9f0bb801)