(m)TLS to PG - #627
Draft
shortishly wants to merge 217 commits into
Draft
Conversation
This change refactors the consumer protocol types (like MemberMetadata, MemberAssignment) from the broker crate into the sans-io crate. These are protocol-level types that belong with the other protocol serialization code. The sans-io crate now handles all encoding/decoding of the consumer protocol types independent of IO concerns.
The commit adds support for consumer groups including: - Client-side Group data structure - Range partition assignor implementation - Consumer protocol (de)serialization - Consumer group assignor trait and types
Bumps [actions/download-artifact](https://github.com/actions/download-artifact) from 7 to 8. - [Release notes](https://github.com/actions/download-artifact/releases) - [Commits](actions/download-artifact@v7...v8) --- updated-dependencies: - dependency-name: actions/download-artifact dependency-version: '8' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
Bumps [actions/upload-artifact](https://github.com/actions/upload-artifact) from 6 to 7. - [Release notes](https://github.com/actions/upload-artifact/releases) - [Commits](actions/upload-artifact@v6...v7) --- updated-dependencies: - dependency-name: actions/upload-artifact dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
Bumps [fake](https://github.com/cksac/fake-rs) from 4.4.0 to 5.1.0. - [Release notes](https://github.com/cksac/fake-rs/releases) - [Commits](https://github.com/cksac/fake-rs/commits) --- updated-dependencies: - dependency-name: fake dependency-version: 5.1.0 dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
Bumps [crate-ci/typos](https://github.com/crate-ci/typos) from 1.44.0 to 1.48.0. - [Release notes](https://github.com/crate-ci/typos/releases) - [Changelog](https://github.com/crate-ci/typos/blob/master/CHANGELOG.md) - [Commits](crate-ci/typos@v1.44.0...v1.48.0) --- updated-dependencies: - dependency-name: crate-ci/typos dependency-version: 1.48.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com>
Bumps [actions/setup-go](https://github.com/actions/setup-go) from 5 to 7. - [Release notes](https://github.com/actions/setup-go/releases) - [Commits](actions/setup-go@v5...v7) --- updated-dependencies: - dependency-name: actions/setup-go dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
Bumps [actions/upload-artifact](https://github.com/actions/upload-artifact) from 4 to 7. - [Release notes](https://github.com/actions/upload-artifact/releases) - [Commits](actions/upload-artifact@v4...v7) --- updated-dependencies: - dependency-name: actions/upload-artifact dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
…actions/upload-artifact-7
…actions/setup-go-7
…crate-ci/typos-1.48.0
…softprops/action-gh-release-3 build(deps): bump softprops/action-gh-release from 2 to 3
build(deps): bump fake from 4.4.0 to 5.1.0
…extractions/setup-just-4 build(deps): bump extractions/setup-just from 3 to 4
Bumps [actions/checkout](https://github.com/actions/checkout) from 4 to 7. - [Release notes](https://github.com/actions/checkout/releases) - [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md) - [Commits](actions/checkout@v4...v7) --- updated-dependencies: - dependency-name: actions/checkout dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
…/actions/checkout-7 build(deps): bump actions/checkout from 4 to 7
- abort transactions whose transaction_timeout_ms has elapsed via the existing end_in_tx path, releasing the last stable offset they pin - guard end_in_tx with a status SELECT ... FOR UPDATE against a racing or retried EndTxn - let txn_detail's timer reset on a resolved (COMMITTED/ABORTED) row, not just an untouched one - txn_detail.started_at becomes timestamptz
- treat PREPARE_COMMIT/PREPARE_ABORT as finalization in progress: a matching retry no-ops without writing a second control marker, a conflicting outcome returns InvalidTxnState instead of acking success - tie txn_detail to the producer epoch in the two delete_by_txn queries, which deleted bookkeeping for every epoch of the producer - order txn_select_produced_topitions deterministically so concurrent finalizations acquire watermark row locks in the same order - never expire a transaction_timeout_ms of 0; cast the sweep's $2 to timestamptz
- maintain_transactions now aborts through abort_timed_out, which bumps the producer's epoch: without it a still-alive producer keeps writing under its old epoch into a transaction the broker already finalized, and that data reaches read_committed consumers as ordinary committed data - end_in_tx validates the request's epoch before touching transaction state, returning ProducerFenced for a stale epoch and UnknownProducerId for an unknown producer
…nicking init_producer had a todo!() for anything that was not literally (-1, -1): API versions 2 and below omit producer_id/epoch entirely and decode as (None, None), and v3+ epoch-bump recovery carries a real id and epoch. The (-1, -1) logic moves to bump_or_create_producer; recovery requests are validated against the current epoch (ProducerFenced when stale), and malformed partial requests get InvalidRequest.
Re-adding a partition already in the transaction violated txn_topition's unique (txn_detail, topition) constraint, surfacing to the client as a broken connection mid-transaction; Kafka treats the repeat add as a no-op.
feat(storage): abort timed-out transactions in postgres
…ions record_fetch withheld any row whose inserting transaction was not older than the global transaction horizon (pg_snapshot_xmin), so a single open transaction anywhere in the database made committed records temporarily invisible to fetch; produce already serializes offsets per partition through the watermark row lock held to commit, so every visible row below the committed high watermark is final and the guard is unneeded.
…-txn fix(storage): stop postgres fetch stalling on unrelated open transactions
topic_delete_by.sql narrowed the delete on topic.cluster rather than topic.id, so the subquery only established that the named topic existed and the delete then matched every topic in the cluster. topition references topic on delete cascade, and record and header cascade from topition, so a single DeleteTopics wiped every topic and every record in the cluster. Caught by librdkafka 0033_regex_subscribe against postgres, which deletes two topics at teardown: the first delete succeeded and took the second with it, so the second returned UNKNOWN_TOPIC_OR_PARTITION. Narrow the delete on topic.id, matching every sibling delete_by_topic statement. The SQL is shared, so this covers the pg, lite and limbo backends.
fix(storage): scope topic delete to the named topic
Kafka record headers are an ordered list that may repeat a key. The header table keyed on (topition, offset_id, k) and header_fetch.sql had no order by, so postgres served headers from that primary key index in key order rather than produce order, and a repeated key could not be stored at all. Caught by librdkafka 0073_headers and 0085_headers against postgres. 0073 produces msgid, static, multi, multi, multi, null, empty and validates by index; sorted by key that starts with "empty", which is what it reported at index 0. 0085 produces header_0..header_260 and got header_10 back at index 2, the byte-sorted position. The existing round-trip coverage in nisshi-broker/tests/fetch.rs compares headers as a BTreeSet, so it could not see either ordering or duplicates -- only the compat suite caught this. Add an ordinal column recording the position of each header within its record, key the table on (topition, offset_id, ordinal) so repeated keys are storable, and order the fetch by it. The DDL and SQL are shared, so this covers the pg, lite and limbo backends. The header table changes shape: an existing database needs to be recreated, or migrated by adding the column and replacing the primary key.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
work in progress