Skip to content

MINOR: Streams broker down flaky test #6041

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 5 commits into from
Dec 20, 2018
Merged

MINOR: Streams broker down flaky test #6041

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e21be80
MINOR: fixes for making test more stable
bbejeck Dec 13, 2018
5e0af05
MINOR: Wait for streams to re-connect with broker
bbejeck Dec 14, 2018
c6a851e
MINOR: Move restart of kafka node inside grabbing the monitor
bbejeck Dec 15, 2018
c9d542e
MINOR: More clean up and close up other timing error gaps
bbejeck Dec 16, 2018
b7cd8aa
MINOR: Clean up messages
bbejeck Dec 16, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions tests/kafkatest/tests/streams/base_streams_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ def get_producer(self, topic, num_messages, repeating_keys=None):
topic,
max_messages=num_messages,
acks=1,
throughput=1000,

@bbejeck bbejeck Dec 17, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

set throughput to 1K per sec keep the volume records down, in some cases where the test runs long I've encountered OOM (default is 100K per sec)

repeating_keys=repeating_keys)

def assert_produce_consume(self,
Expand Down
222 changes: 157 additions & 65 deletions tests/kafkatest/tests/streams/streams_broker_down_resilience_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,9 @@ class StreamsBrokerDownResilience(BaseStreamsTest):
inputTopic = "streamsResilienceSource"
outputTopic = "streamsResilienceSink"
client_id = "streams-broker-resilience-verify-consumer"
num_messages = 5
num_messages = 10000
message = "processed[0-9]*messages"
connected_message = "Discovered group coordinator"

def __init__(self, test_context):
super(StreamsBrokerDownResilience, self).__init__(test_context,
Expand All @@ -48,8 +50,6 @@ def test_streams_resilient_to_broker_down(self):
processor = StreamsBrokerDownResilienceService(self.test_context, self.kafka, self.get_configs())
processor.start()

# until KIP-91 is merged we'll only send 5 messages to assert Kafka Streams is running before taking the broker down
# After KIP-91 is merged we'll continue to send messages the duration of the test
self.assert_produce_consume(self.inputTopic,

@bbejeck bbejeck Dec 17, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not breaking the produce/consume assertion here as there is only one streams client in this test.

self.outputTopic,
self.client_id,
Expand All @@ -61,7 +61,11 @@ def test_streams_resilient_to_broker_down(self):

time.sleep(broker_down_time_in_seconds)

self.kafka.start_node(node)
with processor.node.account.monitor_log(processor.LOG_FILE) as monitor:

@bbejeck bbejeck Dec 17, 2018 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was one source of flakiness as before starting kafka and the subsequent discovery of group coordinator could happen before we can grab monitor and inspect log file. This is done throughout the test, but I'll only comment here.

self.kafka.start_node(node)
monitor.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw output '%s' on " % self.connected_message) + str(processor.node.account))

self.assert_produce_consume(self.inputTopic,
self.outputTopic,
Expand Down Expand Up @@ -95,22 +99,45 @@ def test_streams_runs_with_broker_down_initially(self):
self.wait_for_verification(processor_2, broker_unavailable_message, processor_2.LOG_FILE, 10)
self.wait_for_verification(processor_3, broker_unavailable_message, processor_3.LOG_FILE, 10)

# now start broker
self.kafka.start_node(node)

# assert streams can process when starting with broker down
self.assert_produce_consume(self.inputTopic,
self.outputTopic,
self.client_id,
"running_with_broker_down_initially",
num_messages=9,
timeout_sec=120)

message = "processed3messages"
# need to show all 3 instances processed messages
self.wait_for_verification(processor, message, processor.STDOUT_FILE)
self.wait_for_verification(processor_2, message, processor_2.STDOUT_FILE)
self.wait_for_verification(processor_3, message, processor_3.STDOUT_FILE)
with processor.node.account.monitor_log(processor.LOG_FILE) as monitor_1:

@bbejeck bbejeck Dec 17, 2018 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Grab monitor for each respective streams client before starting kafka

with processor_2.node.account.monitor_log(processor_2.LOG_FILE) as monitor_2:
with processor_3.node.account.monitor_log(processor_3.LOG_FILE) as monitor_3:
self.kafka.start_node(node)

monitor_1.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.connected_message) + str(processor.node.account))
monitor_2.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_2.node.account))
monitor_3.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_3.node.account))

with processor.node.account.monitor_log(processor.STDOUT_FILE) as monitor_1:

@bbejeck bbejeck Dec 17, 2018 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar process here, grab monitor before producing any messages

with processor_2.node.account.monitor_log(processor_2.STDOUT_FILE) as monitor_2:
with processor_3.node.account.monitor_log(processor_3.STDOUT_FILE) as monitor_3:

self.assert_produce(self.inputTopic,

@bbejeck bbejeck Dec 17, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

break assertion of production to streams app and consuming from topic written to by streams into separate parts. I've done this in other parts of the test, but I won't comment on each section.

"sending_message_after_broker_down_initially",
num_messages=self.num_messages,
timeout_sec=120)

monitor_1.wait_until(self.message,

@bbejeck bbejeck Dec 17, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

confirm each streams client is processing messages. Same as above, I've done this in other sections but I'll only call it out here.

timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor.node.account))
monitor_2.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_2.node.account))
monitor_3.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_3.node.account))

self.assert_consume(self.client_id,
"consuming_message_after_broker_down_initially",
self.outputTopic,
num_messages=self.num_messages,
timeout_sec=120)

self.kafka.stop()

Expand All @@ -126,24 +153,40 @@ def test_streams_should_scale_in_while_brokers_down(self):
processor_2.start()

processor_3 = StreamsBrokerDownResilienceService(self.test_context, self.kafka, configs)
processor_3.start()

# need to wait for rebalance once
self.wait_for_verification(processor_3, "State transition from REBALANCING to RUNNING", processor_3.LOG_FILE)

# assert streams can process when starting with broker up
self.assert_produce_consume(self.inputTopic,
self.outputTopic,
self.client_id,
"waiting for rebalance to complete",
num_messages=9,
timeout_sec=120)

message = "processed3messages"

self.wait_for_verification(processor, message, processor.STDOUT_FILE)
self.wait_for_verification(processor_2, message, processor_2.STDOUT_FILE)
self.wait_for_verification(processor_3, message, processor_3.STDOUT_FILE)
rebalance = "State transition from REBALANCING to RUNNING"
with processor_3.node.account.monitor_log(processor_3.LOG_FILE) as monitor:
processor_3.start()

monitor.wait_until(rebalance,
timeout_sec=120,
err_msg=("Never saw output '%s' on " % rebalance) + str(processor_3.node.account))

with processor.node.account.monitor_log(processor.STDOUT_FILE) as monitor_1:
with processor_2.node.account.monitor_log(processor_2.STDOUT_FILE) as monitor_2:
with processor_3.node.account.monitor_log(processor_3.STDOUT_FILE) as monitor_3:

self.assert_produce(self.inputTopic,
"sending_message_normal_broker_start",
num_messages=self.num_messages,
timeout_sec=120)

monitor_1.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor.node.account))
monitor_2.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_2.node.account))
monitor_3.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_3.node.account))

self.assert_consume(self.client_id,
"consuming_message_normal_broker_start",
self.outputTopic,
num_messages=self.num_messages,
timeout_sec=120)

node = self.kafka.leader(self.inputTopic)
self.kafka.stop_node(node)
Expand All @@ -155,17 +198,20 @@ def test_streams_should_scale_in_while_brokers_down(self):
self.wait_for_verification(processor, shutdown_message, processor.STDOUT_FILE)
self.wait_for_verification(processor_2, shutdown_message, processor_2.STDOUT_FILE)

self.kafka.start_node(node)
with processor_3.node.account.monitor_log(processor_3.LOG_FILE) as monitor_3:
self.kafka.start_node(node)

monitor_3.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_3.node.account))

self.assert_produce_consume(self.inputTopic,
self.outputTopic,
self.client_id,
"sending_message_after_stopping_streams_instance_bouncing_broker",
num_messages=9,
num_messages=self.num_messages,
timeout_sec=120)

self.wait_for_verification(processor_3, "processed9messages", processor_3.STDOUT_FILE)

@guozhangwang guozhangwang Dec 17, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is just a cleanup since this line is not necessary, right?

@bbejeck bbejeck Dec 17, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that's correct.


self.kafka.stop()

def test_streams_should_failover_while_brokers_down(self):
Expand All @@ -180,24 +226,40 @@ def test_streams_should_failover_while_brokers_down(self):
processor_2.start()

processor_3 = StreamsBrokerDownResilienceService(self.test_context, self.kafka, configs)
processor_3.start()

# need to wait for rebalance once
self.wait_for_verification(processor_3, "State transition from REBALANCING to RUNNING", processor_3.LOG_FILE)

# assert streams can process when starting with broker up
self.assert_produce_consume(self.inputTopic,
self.outputTopic,
self.client_id,
"waiting for rebalance to complete",
num_messages=9,
timeout_sec=120)

message = "processed3messages"

self.wait_for_verification(processor, message, processor.STDOUT_FILE)
self.wait_for_verification(processor_2, message, processor_2.STDOUT_FILE)
self.wait_for_verification(processor_3, message, processor_3.STDOUT_FILE)
rebalance = "State transition from REBALANCING to RUNNING"

@guozhangwang guozhangwang Dec 17, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why only monitor on processor_3 not the previous two?

@bbejeck bbejeck Dec 17, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since processor 3 is the last streams client to join the group, once it has transitioned from REBALANCE to RUNNING the test is ready to proceed.

with processor_3.node.account.monitor_log(processor_3.LOG_FILE) as monitor:
processor_3.start()

monitor.wait_until(rebalance,
timeout_sec=120,
err_msg=("Never saw output '%s' on " % rebalance) + str(processor_3.node.account))

with processor.node.account.monitor_log(processor.STDOUT_FILE) as monitor_1:
with processor_2.node.account.monitor_log(processor_2.STDOUT_FILE) as monitor_2:
with processor_3.node.account.monitor_log(processor_3.STDOUT_FILE) as monitor_3:

self.assert_produce(self.inputTopic,
"sending_message_after_normal_broker_start",
num_messages=self.num_messages,
timeout_sec=120)

monitor_1.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor.node.account))
monitor_2.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_2.node.account))
monitor_3.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_3.node.account))

self.assert_consume(self.client_id,
"consuming_message_after_normal_broker_start",
self.outputTopic,
num_messages=self.num_messages,
timeout_sec=120)

node = self.kafka.leader(self.inputTopic)
self.kafka.stop_node(node)
Expand All @@ -206,13 +268,43 @@ def test_streams_should_failover_while_brokers_down(self):
processor_2.abortThenRestart()
processor_3.abortThenRestart()

self.kafka.start_node(node)

self.assert_produce_consume(self.inputTopic,
self.outputTopic,
self.client_id,
"sending_message_after_hard_bouncing_streams_instance_bouncing_broker",
num_messages=9,
timeout_sec=120)

with processor.node.account.monitor_log(processor.LOG_FILE) as monitor_1:
with processor_2.node.account.monitor_log(processor_2.LOG_FILE) as monitor_2:
with processor_3.node.account.monitor_log(processor_3.LOG_FILE) as monitor_3:
self.kafka.start_node(node)

monitor_1.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.connected_message) + str(processor.node.account))
monitor_2.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_2.node.account))
monitor_3.wait_until(self.connected_message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_3.node.account))

with processor.node.account.monitor_log(processor.STDOUT_FILE) as monitor_1:
with processor_2.node.account.monitor_log(processor_2.STDOUT_FILE) as monitor_2:
with processor_3.node.account.monitor_log(processor_3.STDOUT_FILE) as monitor_3:

self.assert_produce(self.inputTopic,
"sending_message_after_hard_bouncing_streams_instance_bouncing_broker",
num_messages=self.num_messages,
timeout_sec=120)

monitor_1.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor.node.account))
monitor_2.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_2.node.account))
monitor_3.wait_until(self.message,
timeout_sec=120,
err_msg=("Never saw '%s' on " % self.message) + str(processor_3.node.account))

self.assert_consume(self.client_id,
"consuming_message_after_stopping_streams_instance_bouncing_broker",
self.outputTopic,
num_messages=self.num_messages,
timeout_sec=120)
self.kafka.stop()