Update metadata after creating/deleting a topic
MODIFY
all_tests.sh
all_tests.sh
# WARN: when changing something here, there should probably also be a change in scripts/ci_tests.sh
-mix test --include integration --include consumer_group --include server_0_p_10_and_later --include server_0_p_9_p_0 --include server_0_p_8_p_0
+mix test --include integration --include consumer_group --include server_0_p_10_and_later --include server_0_p_10_p_1 --include server_0_p_9_p_0 --include server_0_p_8_p_0

MODIFY
lib/kafka_ex/server_0_p_10_and_later.ex
lib/kafka_ex/server_0_p_10_and_later.ex
@@ -236,6 +236,8 @@ defmodule KafkaEx.Server0P10AndLater do
{response, %{state | correlation_id: state.correlation_id + 1}}
end
+ state = update_metadata(state)
+
{:reply, response, state}
end
@@ -285,6 +287,8 @@ defmodule KafkaEx.Server0P10AndLater do
{response, %{state | correlation_id: state.correlation_id + 1}}
end
+ state = update_metadata(state)
+
{:reply, response, state}
end

MODIFY
test/integration/server0_p_10_and_later_test.exs
test/integration/server0_p_10_and_later_test.exs
@@ -3,6 +3,7 @@ defmodule KafkaEx.Server0P10P1AndLater.Test do
import TestHelper
@moduletag :server_0_p_10_and_later
+ @num_partitions 10
@tag :create_topic
test "can create a topic" do
@@ -19,11 +20,14 @@ defmodule KafkaEx.Server0P10P1AndLater.Test do
resp = create_topic(name, config)
assert {:topic_already_exists, name} == parse_create_topic_resp(resp)
- wait_for(fn ->
Enum.member?(existing_topics(), name)
- end)
- assert Enum.member?(existing_topics(), name)
+ assert @num_partitions ==
+ KafkaEx.Protocol.Metadata.Response.partitions_for_topic(
+ KafkaEx.metadata(),
+ name
+ )
+ |> Enum.count()
end
@tag :delete_topic
@@ -39,11 +43,6 @@ defmodule KafkaEx.Server0P10P1AndLater.Test do
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
@@ -78,13 +77,13 @@ defmodule KafkaEx.Server0P10P1AndLater.Test do
[
%{
topic: name,
- num_partitions: 10,
+ num_partitions: @num_partitions,
replication_factor: 1,
replica_assignment: [],
config_entries: config
}
],
- timeout: 5_000
+ timeout: 10_000
)
end

Update metadata after creating/deleting a topic