class Kafka::ProduceOperation

def execute
  @instrumenter.instrument("send_messages.producer") do |notification|
    message_count = @buffer.size
    notification[:message_count] = message_count
    begin
      send_buffered_messages
    ensure
      notification[:sent_message_count] = message_count - @buffer.size
    end
  end
end