Add support for DeleteTopics message (API message 20)
MODIFY
lib/kafka_ex.ex
lib/kafka_ex.ex
@@ -24,6 +24,8 @@ defmodule KafkaEx do
alias KafkaEx.Protocol.SyncGroup.Response, as: SyncGroupResponse
alias KafkaEx.Protocol.CreateTopics.Request, as: CreateTopicsRequest
alias KafkaEx.Protocol.CreateTopics.Response, as: CreateTopicsResponse
+ alias KafkaEx.Protocol.DeleteTopics.Request, as: DeleteTopicsRequest
+ alias KafkaEx.Protocol.DeleteTopics.Response, as: DeleteTopicsResponse
alias KafkaEx.Protocol.ApiVersions.Response, as: ApiVersionsResponse
alias KafkaEx.Server
alias KafkaEx.Stream
@@ -618,6 +620,17 @@ defmodule KafkaEx do
Server.call(worker_name, {:create_topics, requests, timeout})
end
+ @doc """
+ Delete topics. Must provide a list of topic names.
+ """
+ @spec delete_topics([DeleteTopicsRequest.t()], Keyword.t()) ::
+ DeleteTopicsResponse.t()
+ def delete_topics(requests, opts \\ []) do
+ worker_name = Keyword.get(opts, :worker_name, Config.default_worker())
+ timeout = Keyword.get(opts, :timeout, 4000)
+ Server.call(worker_name, {:delete_topics, requests, timeout})
+ end
+
# OTP API
def start(_type, _args) do
max_restarts = Application.get_env(:kafka_ex, :max_restarts, 10)

MODIFY
lib/kafka_ex/protocol.ex
lib/kafka_ex/protocol.ex
@@ -14,7 +14,8 @@ defmodule KafkaEx.Protocol do
leave_group: 13,
sync_group: 14,
api_versions: 18,
- create_topics: 19
+ create_topics: 19,
+ delete_topics: 20
}
# DescribeConfigs 32

MODIFY
lib/kafka_ex/protocol/common.ex
lib/kafka_ex/protocol/common.ex
@@ -43,4 +43,28 @@ defmodule KafkaEx.Protocol.Common do
{items, data_after_array} = read_array(num_items - 1, rest, read_one)
{[item | items], data_after_array}
end
+
+ @spec encode_nullable_string(String.t()) :: binary
+ def encode_nullable_string(text) do
+ case text do
+ nil -> <<-1::16-signed>>
+ _ -> encode_string(text)
+ end
+ end
+
+ @spec encode_string(String.t()) :: binary
+ def encode_string(text) do
+ <<byte_size(text)::16-signed, text::binary>>
+ end
+
+ def map_encode(elems, function) do
+ if nil == elems or [] == elems do
+ <<0::32-signed>>
+ else
+ <<length(elems)::32-signed>> <>
+ (elems
+ |> Enum.map(function)
+ |> Enum.reduce(&(&1 <> &2)))
+ end
+ end
end

MODIFY
lib/kafka_ex/protocol/create_topics.ex
lib/kafka_ex/protocol/create_topics.ex
defmodule KafkaEx.Protocol.CreateTopics do
alias KafkaEx.Protocol
+ import KafkaEx.Protocol.Common
@supported_versions_range {0, 0}
@moduledoc """
@@ -126,30 +127,6 @@ defmodule KafkaEx.Protocol.CreateTopics do
encode_nullable_string(config_entry.config_value)
end
- @spec encode_nullable_string(String.t()) :: binary
- defp encode_nullable_string(text) do
- case text do
- nil -> <<-1::16-signed>>
- _ -> encode_string(text)
- end
- end
-
- @spec encode_string(String.t()) :: binary
- defp encode_string(text) do
- <<byte_size(text)::16-signed, text::binary>>
- end
-
- defp map_encode(elems, function) do
- if nil == elems or [] == elems do
- <<0::32-signed>>
- else
- <<length(elems)::32-signed>> <>
- (elems
- |> Enum.map(function)
- |> Enum.reduce(&(&1 <> &2)))
- end
- end
-
@spec parse_response(binary, integer) :: [] | Response.t()
def parse_response(
<<_correlation_id::32-signed, topic_errors_count::32-signed,

ADD
lib/kafka_ex/protocol/delete_topics.ex
+defmodule KafkaEx.Protocol.DeleteTopics do
+ alias KafkaEx.Protocol
+ import KafkaEx.Protocol.Common
+ @supported_versions_range {0, 0}
+
+ @moduledoc """
+ Implementation of the Kafka DeleteTopics request and response APIs
+
+ See: https://kafka.apache.org/protocol.html#The_Messages_DeleteTopics
+ """
+
+ # DeleteTopics Request (Version: 0) => [topics] timeout
+ # topics => STRING
+ # timeout => INT32
+
+ # DeleteTopics Response (Version: 0) => [topic_error_codes]
+ # topic_error_codes => topic error_code
+ # topic => STRING
+ # error_code => INT16
+
+ defmodule Request do
+ @moduledoc false
+ defstruct topics: nil, timeout: nil
+ @type t :: %Request{topics: [String.t()], timeout: integer}
+ end
+
+ defmodule TopicError do
+ @moduledoc false
+ defstruct topic_name: nil, error_code: nil
+ @type t :: %TopicError{topic_name: binary, error_code: atom}
+ end
+
+ defmodule Response do
+ @moduledoc false
+ defstruct topic_errors: nil
+ @type t :: %Response{topic_errors: [TopicError]}
+ end
+
+ def api_version(api_versions) do
+ KafkaEx.ApiVersions.find_api_version(
+ api_versions,
+ :delete_topics,
+ @supported_versions_range
+ )
+ end
+
+ @spec create_request(integer, binary, Request.t(), integer) :: binary
+ def create_request(
+ correlation_id,
+ client_id,
+ delete_topics_request,
+ api_version
+ )
+
+ def create_request(correlation_id, client_id, delete_topics_request, 0) do
+ Protocol.create_request(:delete_topics, correlation_id, client_id) <>
+ encode_topics(delete_topics_request.topics) <>
+ <<delete_topics_request.timeout::32-signed>>
+ end
+
+ @spec encode_topics([String.t()]) :: binary
+ defp encode_topics(topics) do
+ topics |> map_encode(&encode_string/1)
+ end
+
+ @spec parse_response(binary, integer) :: [] | Response.t()
+ def parse_response(
+ <<_correlation_id::32-signed, topic_errors_count::32-signed,
+ topic_errors::binary>>,
+ 0
+ ) do
+ %Response{
+ topic_errors: parse_topic_errors(topic_errors_count, topic_errors)
+ }
+ end
+
+ @spec parse_topic_errors(integer, binary) :: [TopicError.t()]
+ defp parse_topic_errors(0, _), do: []
+
+ defp parse_topic_errors(
+ topic_errors_count,
+ <<topic_name_size::16-signed, topic_name::size(topic_name_size)-binary,
+ error_code::16-signed, rest::binary>>
+ ) do
+ [
+ %TopicError{
+ topic_name: topic_name,
+ error_code: Protocol.error(error_code)
+ }
+ | parse_topic_errors(topic_errors_count - 1, rest)
+ ]
+ end
+end

MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -18,6 +18,7 @@ defmodule KafkaEx.Server do
alias KafkaEx.Protocol.Produce.Request, as: ProduceRequest
alias KafkaEx.Protocol.SyncGroup.Request, as: SyncGroupRequest
alias KafkaEx.Protocol.CreateTopics.Request, as: CreateTopicsRequest
+ alias KafkaEx.Protocol.DeleteTopics.Request, as: DeleteTopicsRequest
alias KafkaEx.Socket
defmodule State do
@@ -212,6 +213,12 @@ defmodule KafkaEx.Server do
state :: State.t()
) :: {:reply, reply, new_state}
when reply: term, new_state: term
+ @callback kafka_delete_topics(
+ [DeleteTopicsRequest.t()],
+ network_timeout :: integer,
+ state :: State.t()
+ ) :: {:reply, reply, new_state}
+ when reply: term, new_state: term
@callback kafka_api_versions(state :: State.t()) :: {:reply, reply, new_state}
when reply: term, new_state: term
@callback kafka_server_update_metadata(state :: State.t()) ::
@@ -334,6 +341,10 @@ defmodule KafkaEx.Server do
kafka_create_topics(requests, network_timeout, state)
end
+ def handle_call({:delete_topics, topics, network_timeout}, _from, state) do
+ kafka_delete_topics(topics, network_timeout, state)
+ end
+
def handle_call({:api_versions}, _from, state) do
kafka_api_versions(state)
end
@@ -857,7 +868,7 @@ defmodule KafkaEx.Server do
"Parse error during #{inspect(module)}.parse_response. Couldn't parse: #{
inspect(response)
}",
- __STACKTRACE__
+ System.stacktrace()
)
end
end

MODIFY
lib/kafka_ex/server_0_p_10_and_later.ex
lib/kafka_ex/server_0_p_10_and_later.ex
@@ -4,6 +4,7 @@ defmodule KafkaEx.Server0P10AndLater do
"""
use KafkaEx.Server
alias KafkaEx.Protocol.CreateTopics
+ alias KafkaEx.Protocol.DeleteTopics
alias KafkaEx.Protocol.ApiVersions
alias KafkaEx.Server0P8P2
alias KafkaEx.Server0P9P0
@@ -179,6 +180,53 @@ defmodule KafkaEx.Server0P10AndLater do
{:reply, response, %{state | correlation_id: state.correlation_id + 1}}
end
+ def kafka_delete_topics(topics, network_timeout, state) do
+ api_version =
+ case DeleteTopics.api_version(state.api_versions) do
+ {:ok, api_version} ->
+ api_version
+
+ _ ->
+ raise "DeleteTopic is not supported in this version of Kafka, or the versions supported by the client do not match the ones supported by the server."
+ end
+
+ main_request =
+ DeleteTopics.create_request(
+ state.correlation_id,
+ @client_id,
+ %DeleteTopics.Request{
+ topics: topics,
+ timeout: network_timeout
+ },
+ api_version
+ )
+
+ broker = state.brokers |> Enum.find(& &1.is_controller)
+
+ {response, state} =
+ case broker do
+ nil ->
+ Logger.log(:error, "Coordinator for topic is not available")
+ {:topic_not_found, state}
+
+ _ ->
+ response =
+ broker
+ |> NetworkClient.send_sync_request(
+ main_request,
+ config_sync_timeout()
+ )
+ |> case do
+ {:error, reason} -> {:error, reason}
+ response -> DeleteTopics.parse_response(response, api_version)
+ end
+
+ {response, %{state | correlation_id: state.correlation_id + 1}}
+ end
+
+ {:reply, response, state}
+ end
+
def kafka_create_topics(requests, network_timeout, state) do
api_version =
case CreateTopics.api_version(state.api_versions) do

MODIFY
lib/kafka_ex/server_0_p_8_p_0.ex
lib/kafka_ex/server_0_p_8_p_0.ex
@@ -15,6 +15,7 @@ defmodule KafkaEx.Server0P8P0 do
{:nowarn_function, kafka_server_offset_commit: 2},
{:nowarn_function, kafka_server_offset_fetch: 2},
{:nowarn_function, kafka_create_topics: 3},
+ {:nowarn_function, kafka_delete_topics: 3},
{:nowarn_function, kafka_api_versions: 1}
]
@@ -94,6 +95,9 @@ defmodule KafkaEx.Server0P8P0 do
def kafka_create_topics(_, _, _state),
do: raise("CreateTopic is not supported in 0.8.0 version of kafka")
+ def kafka_delete_topics(_, _, _state),
+ do: raise("DeleteTopic is not supported in 0.8.0 version of kafka")
+
defp fetch(request, state) do
case network_request(request, Fetch, state) do
{{:error, error}, state_out} -> {error, state_out}

MODIFY
lib/kafka_ex/server_0_p_8_p_2.ex
lib/kafka_ex/server_0_p_8_p_2.ex
@@ -10,6 +10,7 @@ defmodule KafkaEx.Server0P8P2 do
{:nowarn_function, kafka_server_join_group: 3},
{:nowarn_function, kafka_server_leave_group: 3},
{:nowarn_function, kafka_create_topics: 3},
+ {:nowarn_function, kafka_delete_topics: 3},
{:nowarn_function, kafka_api_versions: 1}
]
@@ -199,6 +200,9 @@ defmodule KafkaEx.Server0P8P2 do
def kafka_create_topics(_, _, _state),
do: raise("CreateTopic is not supported in 0.8.2 version of kafka")
+ def kafka_delete_topics(_, _, _state),
+ do: raise("DeleteTopic is not supported in 0.8.2 version of kafka")
+
defp update_consumer_metadata(state),
do: update_consumer_metadata(state, @retry_count, 0)

MODIFY
lib/kafka_ex/server_0_p_9_p_0.ex
lib/kafka_ex/server_0_p_9_p_0.ex
@@ -7,6 +7,7 @@ defmodule KafkaEx.Server0P9P0 do
# these functions aren't implemented for 0.9.0
@dialyzer [
{:nowarn_function, kafka_create_topics: 3},
+ {:nowarn_function, kafka_delete_topics: 3},
{:nowarn_function, kafka_api_versions: 1}
]
@@ -55,6 +56,9 @@ defmodule KafkaEx.Server0P9P0 do
defdelegate kafka_create_topics(requests, network_timeout, state),
to: Server0P8P2
+ defdelegate kafka_delete_topics(requests, network_timeout, state),
+ to: Server0P8P2
+
def kafka_server_init([args]) do
kafka_server_init([args, self()])
end

MODIFY
server.properties
server.properties
@@ -120,3 +120,6 @@ zookeeper.connect=localhost:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000
+
+# Allow deletion of topics
+delete.topic.enable=true

MODIFY
test/integration/server0_p_10_and_later_test.exs
test/integration/server0_p_10_and_later_test.exs
@@ -8,30 +8,46 @@ defmodule KafkaEx.Server0P10P1AndLater.Test do
test "can create a topic" do
name = "create_topic_#{:rand.uniform(2_000_000)}"
- request = %{
- topic: name,
- num_partitions: 10,
- replication_factor: 1,
- replica_assignment: [],
- config_entries: [
+ config = [
%{config_name: "cleanup.policy", config_value: "compact"},
%{config_name: "min.compaction.lag.ms", config_value: "0"}
]
- }
- resp = KafkaEx.create_topics([request], timeout: 2000)
+ resp = create_topic(name, config)
assert {:no_error, name} == parse_create_topic_resp(resp)
- resp = KafkaEx.create_topics([request], timeout: 2000)
+ resp = create_topic(name, config)
assert {:topic_already_exists, name} == parse_create_topic_resp(resp)
wait_for(fn ->
- topics = KafkaEx.metadata().topic_metadatas |> Enum.map(& &1.topic)
- assert Enum.member?(topics, name)
+ Enum.member?(existing_topics(), name)
end)
+
+ assert Enum.member?(existing_topics(), name)
end
- def parse_create_topic_resp(response) do
+ @tag :delete_topic
+ test "can delete a topic" do
+ name = "delete_topic_#{:rand.uniform(2_000_000)}"
+
+ resp = create_topic(name, [])
+ assert {:no_error, name} == parse_create_topic_resp(resp)
+
+ wait_for(fn ->
+ Enum.member?(existing_topics(), name)
+ end)
+
+ resp = KafkaEx.delete_topics([name], timeout: 5_000)
+ assert {:no_error, name} = parse_delete_topic_resp(resp)
+
+ wait_for(fn ->
+ not Enum.member?(existing_topics(), name)
+ end)
+
+ assert not Enum.member?(existing_topics(), name)
+ end
+
+ defp parse_create_topic_resp(response) do
%KafkaEx.Protocol.CreateTopics.Response{
topic_errors: [
%KafkaEx.Protocol.CreateTopics.TopicError{
@@ -43,4 +59,36 @@ defmodule KafkaEx.Server0P10P1AndLater.Test do
{error_code, topic_name}
end
+
+ defp parse_delete_topic_resp(response) do
+ %KafkaEx.Protocol.DeleteTopics.Response{
+ topic_errors: [
+ %KafkaEx.Protocol.DeleteTopics.TopicError{
+ error_code: error_code,
+ topic_name: topic_name
+ }
+ ]
+ } = response
+
+ {error_code, topic_name}
+ end
+
+ defp create_topic(name, config) do
+ KafkaEx.create_topics(
+ [
+ %{
+ topic: name,
+ num_partitions: 10,
+ replication_factor: 1,
+ replica_assignment: [],
+ config_entries: config
+ }
+ ],
+ timeout: 5_000
+ )
+ end
+
+ defp existing_topics() do
+ KafkaEx.metadata().topic_metadatas |> Enum.map(& &1.topic)
+ end
end

Add support for DeleteTopics message (API message 20)