def native_config(opaque=nil)
      Rdkafka::Bindings.rd_kafka_conf_new.tap do |config|
        # Create config
        @config_hash.merge(REQUIRED_CONFIG).each do |key, value|
          error_buffer = FFI::MemoryPointer.from_string(" " * 256)
          result = Rdkafka::Bindings.rd_kafka_conf_set(
            config,
            key.to_s,
            value.to_s,
            error_buffer,
            256
          )
          unless result == :config_ok
            raise ConfigError.new(error_buffer.read_string)
          end
        end

        # Set opaque pointer that's used as a proxy for callbacks
        if opaque
          pointer = ::FFI::Pointer.new(:pointer, opaque.object_id)
          Rdkafka::Bindings.rd_kafka_conf_set_opaque(config, pointer)

          # Store opaque with the pointer as key. We use this approach instead
          # of trying to convert the pointer to a Ruby object because there is
          # no risk of a segfault this way.
          Rdkafka::Config.opaques[pointer.to_i] = opaque
        end

        # Set log callback
        Rdkafka::Bindings.rd_kafka_conf_set_log_cb(config, Rdkafka::Bindings::LogCallback)

        # Set stats callback
        Rdkafka::Bindings.rd_kafka_conf_set_stats_cb(config, Rdkafka::Bindings::StatsCallback)
      end
    end
def native_config(opaque=nil)
      Rdkafka::Bindings.rd_kafka_conf_new.tap do |config|
        # Create config
        @config_hash.merge(REQUIRED_CONFIG).each do |key, value|
          error_buffer = FFI::MemoryPointer.from_string(" " * 256)
          result = Rdkafka::Bindings.rd_kafka_conf_set(
            config,
            key.to_s,
            value.to_s,
            error_buffer,
            256
          )
          unless result == :config_ok
            raise ConfigError.new(error_buffer.read_string)
          end
        end

        # Set opaque pointer that's used as a proxy for callbacks
        if opaque
          pointer = ::FFI::Pointer.new(:pointer, opaque.object_id)
          Rdkafka::Bindings.rd_kafka_conf_set_opaque(config, pointer)

          # Store opaque with the pointer as key. We use this approach instead
          # of trying to convert the pointer to a Ruby object because there is
          # no risk of a segfault this way.
          Rdkafka::Config.opaques[pointer.to_i] = opaque
        end

        # Set log callback
        Rdkafka::Bindings.rd_kafka_conf_set_log_cb(config, Rdkafka::Bindings::LogCallback)

        # Set stats callback
        Rdkafka::Bindings.rd_kafka_conf_set_stats_cb(config, Rdkafka::Bindings::StatsCallback)
      end
    end
This method is only intented to be used to create a client,
 using it in another way will leak memory.