def handle_crashed_listener(listener, error, retry_count)
      backoff = listener.create_exponential_backoff
      interval = backoff.interval_at(retry_count).round(2)

      metadata = {
        listener_id: listener.id,
        retry_count: retry_count,
        waiting_time: interval
      }.merge(error_metadata(error))

      instrument('executor.retry_listener_error', metadata) do
        log_error("Listener crashed, waiting #{interval}s (#{error.message})", metadata)
        sleep interval
      end
    end
def handle_crashed_listener(listener, error, retry_count)
      backoff = listener.create_exponential_backoff
      interval = backoff.interval_at(retry_count).round(2)

      metadata = {
        listener_id: listener.id,
        retry_count: retry_count,
        waiting_time: interval
      }.merge(error_metadata(error))

      instrument('executor.retry_listener_error', metadata) do
        log_error("Listener crashed, waiting #{interval}s (#{error.message})", metadata)
        sleep interval
      end
    end
rubocop:enable Lint/RescueException
 When "listener#start" is interrupted it's safe to assume that the consumer
 and the kafka client were properly stopped, it's safe to call start
 again