KAFKA-20896: avoid epoch bump when a new config adding to a assignor

[gabriellefu]

change:
add a default map and only trigger a epoch bump for the assignor if it's change from not having the config to default value
added tests

Reviewers: Matthias J. Sax matthias@confluent.io

[gabriellefu]

The default of a config should be something like 0 or null if it's newly added to a assignor in the future version. In this case, if one config is newly added and not set, it should looks the same after withoutDefaults() when it's the first time added to the config.

[mjsax]

Claude review:

The comparison logic is sound — I walked group-level vs. broker-level default combinations in both directions and the symmetric strip never suppresses a real change. And it correctly does not change what the assignor sees, which is the right call with pluggable assignors in the picture.

Findings, most important first:

1. The javadoc added to streamsGroupAssignmentConfigs is false, and acting on it breaks the assignor. It now claims "A configuration is only included once it is set to a non-default value" — but the body is unchanged and num.standby.replicas is still emitted unconditionally (GroupMetadataManager.java:9917). That's not just a stale comment; it's an invitation to the obvious follow-up, and the follow-up crashes:

// StickyTaskAssignor.java:98-100
localState.numStandbyReplicas =
    groupSpec.configs().isEmpty() ? 0
        : Integer.parseInt(groupSpec.configs().get("num.standby.replicas"));

A map that is non-empty but lacks the key — e.g. {rack.aware.assignment.tags: "zone"} once standbys are omitted at default — gives parseInt(null) → NumberFormatException. The invariant keeping that unreachable is precisely the unconditional write the JIRA wants removed. Either revert the javadoc to describe what the method does, or make it do what the javadoc says and fix line 98-100 first.

2. The footgun is relocated, not removed. The JIRA's complaint is that #22213 fixed one config ad hoc and the next one hits it again. After this PR, adding a config still requires remembering a second, unenforced step — registering it in ASSIGNMENT_CONFIG_DEFAULTS — and forgetting it fails silently with no test failure. Cheap structural guard, worth asking for:

assertEquals(Map.of(), withoutDefaults(streamsGroupAssignmentConfigs(groupId))); // all-default broker + group config

That turns "remember to register the default" into a compile-time-adjacent failure. Supporting evidence that the omission is easy to make: StreamsGroupStaticMemberGroupMetadataManagerTest has to seed getDefaultAssignmentConfigs() into ~12 groups purely to dodge the spurious bump.

3. The new test asserts against a state no broker version can produce. withLastAssignmentConfigs(Map.of("rack.aware.assignment.tags", "")) with the comment "A version that did not omit default-valued configurations recorded the tags at their default". The test does pass, and it does exercise the fix, but only via a side effect: the stored map also lacks num.standby.replicas, so it's the standby-replicas strip doing the work, not the tags strip. Name and narration point at the wrong config. The reachable case is worth testing directly: stored map without a config that current emits at its default.

4. Two mechanisms for one rule. rack.aware.assignment.tags is now handled both by the isEmpty() guard at the source and by the defaults map at comparison time. Keeping the source guard is right (dropping it would start writing tags: "" into every group's record and into the assignor's map), but the relationship should be spelled out, otherwise the next person picks one at random.

Nits: withoutDefaults returns a TreeMap — ordering buys nothing for an equals() comparison; the "num.standby.replicas" / "rack.aware.assignment.tags" literals are now duplicated across two places where a typo silently disables the filter, so shared constants would help; and the new heartbeat(...) helper is dropped between two @test methods rather than with the other helpers.

[mjsax]

Another round from Claude. The first one is subtle, and I am frankly not sure about it -- if the logic is not too complex, and we add a proper comment why we need to treat num.standby.replicas differently (ie, 4.3/4.2 downgrade compatibility), it might indeed to worth fixing it.

I flagged the consequence last time but didn't spell out the mechanism, so here it is. Streams groups are GA since 4.2 (StreamsVersion.SV_1, LATEST_PRODUCTION, IBP_4_2_IV1), so a 4.3 → 4.4 rolling upgrade with live streams groups is ordinary. 4.2 and 4.3 write {num.standby.replicas: N}; trunk today writes byte-identically for any group not using tags, because of the guard she removed. After this PR the new broker writes {num.standby.replicas: "0", rack.aware.assignment.tags: ""}. replay reads the key/value list generically (GroupMetadataManager.java:6236), so when a group's coordinator moves to a not-yet-upgraded broker, that broker loads tags: "", computes {nsr:"0"}, sees a difference, and bumps the epoch. The return trip is absorbed by withoutDefaults, so it's one spurious rebalance per group per new→old coordinator move during the rolling window — where trunk today has zero. That's the exact failure mode the JIRA exists to prevent, reintroduced in the mixed-version window.

The rule that avoids it is a bit subtle but worth writing down, because "omit everything at default" is also wrong — omitting num.standby.replicas would make the old broker see {} vs its {nsr:"0"} and bump for the opposite reason. What actually keeps things clean is: the written map should stay byte-identical to what the previous release writes for a group that sets nothing. Concretely — keep num.standby.replicas unconditional because 4.2/4.3 already write it, and omit newly added configs at their default. withoutDefaults stays as the general mechanism absorbing legacy maps, and the guard test still protects the unconditional style for anyone who writes one.

Fair counter-argument for her: it's one bump, transient, and uniformity is genuinely easier to reason about. Worth raising as a question rather than a blocker — and Lucas may have a view on how much mixed-version rebalance churn is acceptable here.

Smaller, new in this revision: the GroupSpec.configs() javadoc now promises implementers that an absent config is at its default — but the reference implementation doesn't honor it. StickyTaskAssignor.java:98-100 still does configs().isEmpty() ? 0 : Integer.parseInt(configs().get("num.standby.replicas")), so a non-empty map missing that key throws NumberFormatException instead of defaulting. Unreachable today, but it's a public contract statement for KIP-1357 assignors that the built-in one breaks; getOrDefault(..., "0") is the whole fix.

Minor: the tags value now reaches assignors as "" rather than being absent — whoever wires it up will find "".split(",") yields [""], one empty tag, which the old omit-when-empty shape avoided. And still: withoutDefaults returns a TreeMap where ordering buys nothing, and heartbeat(...) sits between two @Test methods.

[gabriellefu]

updated the pr base on the comment, if one config is not set, it will not be written in __consumer_offset to avoid if a downgraded broker become the broker and don't recognize the config. @mjsax