feat: kafka protobuf backend improvements and cleanup

[tommy-ca]

Summary

Add Binance/Tardis parity fields to normalized v2beta1 schemas (trade and order book) while keeping wire compatibility. Documentation now tracks the change in a new schema changelog.

What changed

• Trade (v2beta1): add optional maker, event_time, match_id, liquidity_flag for venue parity.
• Order book (v2beta1): add optional event_time, last_update_id for venue parity.
• Mapping docs updated; new docs/schemas/CHANGELOG.md entry for v3.2.0 (unpublished).
• Regenerated buf artifacts (git-ignored) and refreshed descriptors.

Compatibility

• Additive only; no field renames/removals. buf breaking --against buf.build/tommyk/crypto-market-data:main passes.
• Decimal scale remains 1e-8. timestamp still represents match time; event_time is provided when venues expose it.

Testing

• buf lint proto
• buf breaking proto --against buf.build/tommyk/crypto-market-data:main
• buf generate proto; buf export proto; buf build proto -o gen/protobuf/cryptofeed/normalized/v2beta1/descriptor.bin

Next steps

• Publish schemas as v3.2.0 once approved (buf push proto --label main --label v3.2.0 --label latest).

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Make Kafka callback callable via feedhandler

Feed callbacks are invoked with await cb(obj, receipt_timestamp) (see Feed.callback), which calls the inherited BackendCallback.__call__ and expects a write method. The new KafkaBackendBase/KafkaCallback hierarchy added in this commit does not implement write or override __call__, so calling a Kafka callback instance will raise an AttributeError on the first message rather than enqueueing it. This regresses the standard KafkaCallback(... ) usage path—any feed using it will crash immediately instead of producing to Kafka until a callable entry point is provided (e.g., override __call__ to queue messages or add a write implementation).

Useful? React with 👍 / 👎.

[tommy-ca]

Code review

Found 1 issue:

1. Missing task_done() calls in _drain_batch() causing task queue deadlock (bug violates asyncio.Queue contract)

The _drain_batch() method calls queue.get_nowait() but never calls queue.task_done(), violating the asyncio.Queue contract. This will cause task queue deadlock and task leaks in long-running Kafka producers. The bug occurs 100% of the time when batch draining is enabled (default configuration). In contrast, _drain_once() correctly uses try/finally to ensure task_done() is always called.

cryptofeed/cryptofeed/backends/kafka/base.py

Lines 198 to 217 in 9a35b55

	max_batch = self._batch_drain_size
	while batch_count < max_batch:
	try:
	message = self._queue.get_nowait()
	except asyncio.QueueEmpty:
	break
	if message is _STOP_SENTINEL:
	self._running = False
	return
	await self._process_message(message)
	batch_count += 1
	await asyncio.sleep(0)
	# ------------------------------------------------------------------ #
	# Extension points
	# ------------------------------------------------------------------ #

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}