def parse_opts(opts)
      REQUIRED_OPTS.each do |opt|
        raise "#{opt.inspect} option not provided" if opts[opt].nil?
      end

      rafka_opts = opts.reject { |k| k == :redis || k == :librdkafka }

      redis_opts = REDIS_DEFAULTS.dup.merge(opts[:redis] || {})
      redis_opts.merge!(
        rafka_opts.select { |k| [:host, :port, :id].include?(k) }
      )

      [rafka_opts, redis_opts]
    end
def parse_opts(opts)
      REQUIRED_OPTS.each do |opt|
        raise "#{opt.inspect} option not provided" if opts[opt].nil?
      end

      rafka_opts = opts.reject { |k| k == :redis || k == :librdkafka }

      redis_opts = REDIS_DEFAULTS.dup.merge(opts[:redis] || {})
      redis_opts.merge!(
        rafka_opts.select { |k| [:host, :port, :id].include?(k) }
      )

      [rafka_opts, redis_opts]
    end
@param opts [Hash] options hash as passed to {#initialize}

 @return [Array<Hash, Hash>] rafka opts, redis opts