perf(consumers): Route messages based on kafka headers

[nikhars]

Context

The current errors and transactions consumers determine whether the message should be processed or not at a stage when the messages have been deserialized. For example, in transactions consumer it happens at https://github.com/getsentry/snuba/blob/master/snuba/datasets/transactions_processor.py#L78-L79. It is done that way since error and transaction messages are on the same topic.

Improvement

The change introduces a new message pre filter which looks at kafka headers and decides whether the message should be dropped or not.

With this getsentry/sentry#29618 merged, there is a way to do the filtering based on kafka headers. This should improve performance since the consumers don't need to deserialize the message anymore to
determine whether to drop message or not.

Tests

Added unit tests for the KafkaHeaderFilter. Also ran code with some additional logs and saw that the messages being dropped by errors and transactions consumer were mutually exclusive and atleast one consumer was dropping every offset message

00:18:53 transaction-consumer | 2021-10-29 00:18:53,784 Message with offset 21886 dropped 00:18:53 transaction-consumer | 2021-10-29 00:18:53,784 Message with offset 21887 dropped 00:18:53 consumer | 2021-10-29 00:18:53,797 Message with offset 21888 dropped 00:18:53 transaction-consumer | 2021-10-29 00:18:53,894 Message with offset 21889 dropped 00:18:53 transaction-consumer | 2021-10-29 00:18:53,901 Message with offset 21890 dropped 00:18:53 transaction-consumer | 2021-10-29 00:18:53,912 Message with offset 21891 dropped 00:18:53 consumer | 2021-10-29 00:18:53,917 Message with offset 21892 dropped 00:18:54 consumer | 2021-10-29 00:18:54,035 Message with offset 21893 dropped 00:18:54 transaction-consumer | 2021-10-29 00:18:54,063 Message with offset 21894 dropped 00:18:54 transaction-consumer | 2021-10-29 00:18:54,081 Message with offset 21895 dropped 00:18:54 consumer | 2021-10-29 00:18:54,174 Message with offset 21896 dropped 00:18:54 transaction-consumer | 2021-10-29 00:18:54,234 Message with offset 21897 dropped 00:18:54 consumer | 2021-10-29 00:18:54,284 Message with offset 21898 dropped 00:18:54 transaction-consumer | 2021-10-29 00:18:54,324 Message with offset 21899 dropped

[evanh]

I assume that it's OK if both consumers still process the message? For example if the header is missing or has a malformed value?

[nikhars]

I assume that it's OK if both consumers still process the message? For example if the header is missing or has a malformed value?

Yes, its OK if both consumers process the message. In that case the existing errors/transactions processors still have the check to process message relevant to them.

[codecov-commenter]

Codecov Report

Merging #2176 (f00b904) into master (f664b7a) will increase coverage by 0.00%.
The diff coverage is 100.00%.

@@           Coverage Diff           @@
##           master    #2176   +/-   ##
=======================================
  Coverage   91.17%   91.18%           
=======================================
  Files         546      547    +1     
  Lines       24279    24301   +22     
=======================================
+ Hits        22136    22158   +22     
  Misses       2143     2143           

Impacted Files	Coverage Δ
snuba/datasets/message_filters.py	91.30% <100.00%> (+7.97%)	⬆️
snuba/datasets/storages/errors.py	100.00% <100.00%> (ø)
snuba/datasets/storages/transactions.py	100.00% <100.00%> (ø)
tests/datasets/test_message_filters.py	100.00% <100.00%> (ø)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f664b7a...f00b904. Read the comment docs.