Merge pull request #530 from nguyenquangminh0711/describe-configs-api

Describe topic API

Author: dasch

.circleci/config.yml

@@ -58,8 +58,6 @@ jobs:
           KAFKA_DELETE_TOPIC_ENABLE: true
       - image: wurstmeister/kafka:0.11.0.1
         environment:
-          KAFKA_ADVERTISED_HOST_NAME: localhost
-          KAFKA_ADVERTISED_PORT: 9093
           KAFKA_PORT: 9093
           KAFKA_ZOOKEEPER_CONNECT: localhost:2181
           KAFKA_DELETE_TOPIC_ENABLE: true

lib/kafka/broker.rb

@@ -121,6 +121,12 @@ def delete_topics(**options)
       send_request(request)
     end
 
+    def describe_configs(**options)
+      request = Protocol::DescribeConfigsRequest.new(**options)
+
+      send_request(request)
+    end
+
     def create_partitions(**options)
       request = Protocol::CreatePartitionsRequest.new(**options)
 

lib/kafka/client.rb

@@ -473,6 +473,15 @@ def delete_topic(name, timeout: 30)
       @cluster.delete_topic(name, timeout: timeout)
     end
 
+    # Describe configs of a topic.
+    #
+    # @param name [String] the name of the topic.
+    # @param configs [Array<String>] array of desired config names.
+    # @return [Hash<string, string>]
+    def describe_topic(name, configs = [])
+      @cluster.describe_topic(name, configs)
+    end
+
     # Create partitions for a topic.
     #
     # @param name [String] the name of the topic.

lib/kafka/cluster.rb

@@ -210,6 +210,25 @@ def delete_topic(name, timeout:)
       @logger.info "Topic `#{name}` was deleted"
     end
 
+    def describe_topic(name, configs = [])
+      options = {
+        resources: [[Kafka::Protocol::RESOURCE_TYPE_TOPIC, name, configs]]
+      }
+      broker = controller_broker
+
+      @logger.info "Fetching topic `#{name}`'s configs using controller broker #{broker}"
+
+      response = broker.describe_configs(**options)
+
+      response.resources.each do |resource|
+        Protocol.handle_error(resource.error_code, resource.error_message)
+      end
+      topic_description = response.resources.first
+      topic_description.configs.each_with_object({}) do |config, hash|
+        hash[config.name] = config.value
+      end
+    end
+
     def create_partitions_for(name, num_partitions:, timeout:)
       options = {
         topics: [[name, num_partitions, nil]],

lib/kafka/protocol.rb

@@ -27,6 +27,7 @@ module Protocol
     API_VERSIONS_API = 18
     CREATE_TOPICS_API = 19
     DELETE_TOPICS_API = 20
+    DESCRIBE_CONFIGS_API = 32
     CREATE_PARTITIONS_API = 37
 
     # A mapping from numeric API keys to symbolic API names.
@@ -46,6 +47,7 @@ module Protocol
       API_VERSIONS_API => :api_versions,
       CREATE_TOPICS_API => :create_topics,
       DELETE_TOPICS_API => :delete_topics,
+      DESCRIBE_CONFIGS_API => :describe_configs_api,
       CREATE_PARTITIONS_API => :create_partitions
     }
 
@@ -91,6 +93,25 @@ module Protocol
       42 => InvalidRequest
     }
 
+    # A mapping from int to corresponding resource type in symbol.
+    # https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java
+    RESOURCE_TYPE_UNKNOWN = 0
+    RESOURCE_TYPE_ANY = 1
+    RESOURCE_TYPE_TOPIC = 2
+    RESOURCE_TYPE_GROUP = 3
+    RESOURCE_TYPE_CLUSTER = 4
+    RESOURCE_TYPE_TRANSACTIONAL_ID = 5
+    RESOURCE_TYPE_DELEGATION_TOKEN = 6
+    RESOURCE_TYPES = {
+      RESOURCE_TYPE_UNKNOWN => :unknown,
+      RESOURCE_TYPE_ANY => :any,
+      RESOURCE_TYPE_TOPIC => :topic,
+      RESOURCE_TYPE_GROUP => :group,
+      RESOURCE_TYPE_CLUSTER => :cluster,
+      RESOURCE_TYPE_TRANSACTIONAL_ID => :transactional_id,
+      RESOURCE_TYPE_DELEGATION_TOKEN => :delegation_token,
+    }
+
     # Handles an error code by either doing nothing (if there was no error) or
     # by raising an appropriate exception.
     #
@@ -147,5 +168,7 @@ def self.api_name(api_key)
 require "kafka/protocol/create_topics_response"
 require "kafka/protocol/delete_topics_request"
 require "kafka/protocol/delete_topics_response"
+require "kafka/protocol/describe_configs_request"
+require "kafka/protocol/describe_configs_response"
 require "kafka/protocol/create_partitions_request"
 require "kafka/protocol/create_partitions_response"

lib/kafka/protocol/describe_configs_request.rb

@@ -0,0 +1,33 @@
+module Kafka
+  module Protocol
+
+    class DescribeConfigsRequest
+      def initialize(resources:)
+        @resources = resources
+      end
+
+      def api_key
+        DESCRIBE_CONFIGS_API
+      end
+
+      def api_version
+        0
+      end
+
+      def response_class
+        Protocol::DescribeConfigsResponse
+      end
+
+      def encode(encoder)
+        encoder.write_array(@resources) do |type, name, configs|
+          encoder.write_int8(type)
+          encoder.write_string(name)
+          encoder.write_array(configs) do |config|
+            encoder.write_string(config)
+          end
+        end
+      end
+    end
+
+  end
+end

lib/kafka/protocol/describe_configs_response.rb

@@ -0,0 +1,71 @@
+module Kafka
+  module Protocol
+    class DescribeConfigsResponse
+      class ResourceDescription
+        attr_reader :name, :type, :error_code, :error_message, :configs
+
+        def initialize(name:, type:, error_code:, error_message:, configs:)
+          @name = name
+          @type = type
+          @error_code = error_code
+          @error_message = error_message
+          @configs = configs
+        end
+      end
+
+      class ConfigEntry
+        attr_reader :name, :value, :read_only, :is_default, :is_sensitive
+
+        def initialize(name:, value:, read_only:, is_default:, is_sensitive:)
+          @name = name
+          @value = value
+          @read_only = read_only
+          @is_default = is_default
+          @is_sensitive = is_sensitive
+        end
+      end
+
+      attr_reader :resources
+
+      def initialize(throttle_time_ms:, resources:)
+        @throttle_time_ms = throttle_time_ms
+        @resources = resources
+      end
+
+      def self.decode(decoder)
+        throttle_time_ms = decoder.int32
+        resources = decoder.array do
+          error_code = decoder.int16
+          error_message = decoder.string
+
+          resource_type = decoder.int8
+          if Kafka::Protocol::RESOURCE_TYPES[resource_type].nil?
+            raise Kafka::ProtocolError, "Resource type not supported: #{resource_type}"
+          end
+          resource_name = decoder.string
+
+          configs = decoder.array do
+            ConfigEntry.new(
+              name: decoder.string,
+              value: decoder.string,
+              read_only: decoder.boolean,
+              is_default: decoder.boolean,
+              is_sensitive: decoder.boolean,
+            )
+          end
+
+          ResourceDescription.new(
+            type: RESOURCE_TYPES[resource_type],
+            name: resource_name,
+            error_code: error_code,
+            error_message: error_message,
+            configs: configs
+          )
+        end
+
+        new(throttle_time_ms: throttle_time_ms, resources: resources)
+      end
+    end
+
+  end
+end

spec/functional/topic_management_spec.rb

@@ -32,4 +32,17 @@
     kafka.create_partitions_for(topic, num_partitions: 10)
     expect(kafka.partitions_for(topic)).to eq 10
   end
+
+  example "describe a topic" do
+    unless kafka.supports_api?(Kafka::Protocol::DESCRIBE_CONFIGS_API)
+      skip("This Kafka version not support ")
+    end
+
+    topic = generate_topic_name
+    configs = kafka.describe_topic(topic, %w(retention.ms retention.bytes non_exists))
+
+    expect(configs.keys).to eql(%w(retention.ms retention.bytes))
+    expect(configs['retention.ms']).to be_a(String)
+    expect(configs['retention.bytes']).to be_a(String)
+  end
 end