def query_watermark_offsets(topic, partition, timeout_ms=200)
      low = FFI::MemoryPointer.new(:int64, 1)
      high = FFI::MemoryPointer.new(:int64, 1)

      response = Rdkafka::Bindings.rd_kafka_query_watermark_offsets(
        @native_kafka,
        topic,
        partition,
        low,
        high,
        timeout_ms
      )
      if response != 0
        raise Rdkafka::RdkafkaError.new(response, "Error querying watermark offsets for partition #{partition} of #{topic}")
      end

      return low.read_int64, high.read_int64
    end
def query_watermark_offsets(topic, partition, timeout_ms=200)
      low = FFI::MemoryPointer.new(:int64, 1)
      high = FFI::MemoryPointer.new(:int64, 1)

      response = Rdkafka::Bindings.rd_kafka_query_watermark_offsets(
        @native_kafka,
        topic,
        partition,
        low,
        high,
        timeout_ms
      )
      if response != 0
        raise Rdkafka::RdkafkaError.new(response, "Error querying watermark offsets for partition #{partition} of #{topic}")
      end

      return low.read_int64, high.read_int64
    end
Query broker for low (oldest/beginning) and high (newest/end) offsets for a partition.

 @param topic [String] The topic to query
 @param partition [Integer] The partition to query
 @param timeout_ms [Integer] The timeout for querying the broker

 @raise [RdkafkaError] When querying the broker fails.

 @return [Integer] The low and high watermark