feat: Add rebalance integration testing via pytest and setup integration testing on CI

[enochtangg]

Overview

This is PR is mainly responsible for adding an end-to-end test for validating that consumer rebalances should not produce 1 or more messages in sqlite. More importantly, this PR introduces python dependencies (pytest) into this project as it is much simpler/quicker to write these end-to-end test in python.

Testing

Run make integration-test

[enochtangg]

Looking for some feedback on the following:

1. Thoughts on using pytest for end-to-end integration testing with sentry?
2. Currently, the test_rebalancing_only_processed_once() doesn't actually produce any messages. One thought was to run the send_task CLI command as a previous step in CI before running the pytest. But this wouldn't be repeatable for other tests. Maybe we could run the CLI command in the python test?
3. The test could probably use some clean up.

[evanh]

I'm not sure I understand the purpose of this function. At the moment it starts, and then after a random delay dies. Is that the intended behaviour? There's no guarantee it will process the messages assigned to it. I think Mark had a good idea in his PR, where he passes in the number of expected messages, and can track to see whether the consumer has processed them or not.

[enochtangg]

When the function starts, it executes the rust binary in a new child process. Then, thread itself sleeps for a random delay. This gives the consumer a chance to start up, receive assigned partitions, and process messages. Finally, a SIGINT is sent to the process and the consumer shutsdown gracefully.

It is true that we aren't explicitly checking all messages are processed here. It is possible that the consumer is started and dies before it can process any messages (though unlikely given the min_sleep and max_sleep parameters). The purpose of the test is to check for duplicate messages during rebalancing, not necessarily that all messages have been processed.

[markstory]

You don't need to set the seed multiple times. Doing it once before you generate other random values is enough.

[enochtangg]

One of the outcomes of seeding, is that all (8) consumers restart at the same time for every iteration. FWIW, I think this is also a good test to have because we'd want to simulate both rolling deploys and deploys which shutdown multiple consumers at the same time.

[markstory]

How do we fixate the time/seed when trying to reproduce a failure? We could read from an environment variable like TEST_SEED 🤷

[john-z-yang]

What if we just have the seed as cli argument? And in our gha it's always set to 1 or something
Ooops never mind, I see we already went for envar

[enochtangg]

This integration test no longer relies on sentry to produce test messages for integration tests for two reasons:

1. It's nicer to separate the two concerns and functionally test each component without potentially breaking each other. Breaking sentry taskworker will not break taskbroker CI.
2. With the new devinfra tool, we might not need to clone and install deps for sentry in taskbroker CI. This removes that requirement.

[markstory]

Looking good. I think we should move forward with this. We can refine and expand it incrementally.

[enochtangg]

After a long investigation, @john-z-yang and I have found what change was causing CI to fail this integration test. Strangely enough, this test passed consistently locally which made investigating this troublesome. When this test was executed on ubuntu, the consumer would write duplicate messages (typically 0-2 before and after the rebalance). Reverting this change: 2b9064c#diff-316edec7494fa460e0a4bdc62a604da6b4caa7933db418d023bbfd27c6277c4dR64 fixed the problem. This is because when using select! macro, the consumer client shutdown step was not fully completed since the check_for_shutdown() yields first during rebalancing. As a result, the commit state was not being saved correctly causing duplicate entries.

[evanh]

I don't think this is now working the way we want: how to handle_events and handle_consumer_client signal shutdowns if they error out? I think if those threads just stop running, the consumer won't shut down?

[john-z-yang]

Do you mind elaborating on this? I'm having a hard time coming up with a scenario where those 2 threads panic