mirror of
https://github.com/hanzoai/kafka.git
synced 2026-10-30 14:48:25 +00:00
zeekayandHanzo Dev
e58c08bce5
The 2026-07 insights outage, decomplected into its three real defects: - CommitOffset rejected any offset above 2^50 as 'corruption' and GetCommittedOffset purged stored ones. Hanzo PubSub allocates sequences from a sparse e18 space, so every legitimate commit was refused (librdkafka's 'Unknown broker error') and group state self-destructed. Offsets are opaque int64s now; only negative is invalid. - Anything published raw onto a kafka-* subject was served verbatim as a record batch; the client read its bytes 8-11 as a batch length and died with MSG_SIZE_TOO_LARGE, produce died parsing it, and binary search went blind — one stray message wedged the partition until a stream purge. walkBatches() is now the single gate: produce refuses chains that do not walk (INVALID_RECORD), fetch and bounds skip stored messages that do not. - Sequence arithmetic assumed density. All log positions now come from the stamped batch headers via partitionBounds/findRecordSet, reading through NextMessage (by-start-sequence subscription) so holes and e18-sparse sequence spaces behave identically to dense ones. Also: fetch answers OFFSET_OUT_OF_RANGE instead of stalling consumers on dead positions; per-partition errors replace the top-level goto; the response encoder grows to fit instead of silently truncating past 64KiB; request frames are capped at 128MiB so a port scan cannot OOM the broker; OffsetFetch v8 no longer misparses (member fields are v9+) and its response drops a field the schema never had. Co-authored-by: Hanzo Dev <[email protected]>
25 lines
863 B
Go
25 lines
863 B
Go
package protocol
|
|
|
|
import (
|
|
"testing"
|
|
|
|
"github.com/hanzoai/kafka/types"
|
|
)
|
|
|
|
// TestServeReturnsErrorOnUnreachablePubSub proves the embed-safety property: on
|
|
// a startup failure Serve RETURNS an error rather than calling log.Panic /
|
|
// os.Exit, so a host binary (hanzoai/cloud) can run the adaptor in-process
|
|
// without the whole process dying. pubsub.NewClient wraps nats.Connect, which
|
|
// fails fast against an unreachable URL, so Serve returns before it ever binds
|
|
// the Kafka port or starts the admin server (AdminPort 0 disables it anyway).
|
|
func TestServeReturnsErrorOnUnreachablePubSub(t *testing.T) {
|
|
b := NewBroker(&types.Configuration{
|
|
PubSubUrl: "nats://127.0.0.1:1", // nothing listens on port 1
|
|
BrokerPort: 0,
|
|
AdminPort: 0,
|
|
})
|
|
if err := b.Serve(); err == nil {
|
|
t.Fatal("Serve returned nil for an unreachable PubSub; want an error")
|
|
}
|
|
}
|