Skip to content

Commit

Permalink
Fix multiple [Producer name] tags on failures
Browse files Browse the repository at this point in the history
On `rescue Kafka::Error` we're rerunning the whole method block
including `@logger.push_tags(@producer.to_s)`, but clearing tags
only in `ensure` which is not run on `retry`
  • Loading branch information
ojab committed Jul 30, 2021
1 parent 8a18f9e commit 9121246
Showing 1 changed file with 36 additions and 34 deletions.
70 changes: 36 additions & 34 deletions lib/kafka/async_producer.rb
Original file line number Diff line number Diff line change
Expand Up @@ -213,44 +213,46 @@ def run
@logger.info "Starting async producer in the background..."

loop do
operation, payload = @queue.pop

case operation
when :produce
produce(payload[0], **payload[1])
deliver_messages if threshold_reached?
when :deliver_messages
deliver_messages
when :shutdown
begin
# Deliver any pending messages first.
@producer.deliver_messages
rescue Error => e
@logger.error("Failed to deliver messages during shutdown: #{e.message}")

@instrumenter.instrument("drop_messages.async_producer", {
message_count: @producer.buffer_size + @queue.size,
})
begin
operation, payload = @queue.pop

case operation
when :produce
produce(payload[0], **payload[1])
deliver_messages if threshold_reached?
when :deliver_messages
deliver_messages
when :shutdown
begin
# Deliver any pending messages first.
@producer.deliver_messages
rescue Error => e
@logger.error("Failed to deliver messages during shutdown: #{e.message}")

@instrumenter.instrument("drop_messages.async_producer", {
message_count: @producer.buffer_size + @queue.size,
})
end

# Stop the run loop.
break
else
raise "Unknown operation #{operation.inspect}"
end
rescue Kafka::Error => e
@logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
@logger.info "Restarting in 10 seconds..."

# Stop the run loop.
break
else
raise "Unknown operation #{operation.inspect}"
sleep 10
retry
rescue Exception => e
@logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
@logger.error "Async producer crashed!"
ensure
@producer.shutdown
@logger.pop_tags
end
end
rescue Kafka::Error => e
@logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
@logger.info "Restarting in 10 seconds..."

sleep 10
retry
rescue Exception => e
@logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
@logger.error "Async producer crashed!"
ensure
@producer.shutdown
@logger.pop_tags
end

private
Expand Down

0 comments on commit 9121246

Please sign in to comment.