Rewrite TransactionsBounceTest from Scala to Java and move it to
clients-integration-tests. The transaction status header helpers are
moved to ClientsTestUtils and shared with ProducerIdExpirationTest, and
the Scala helpers in TestUtils that were only used by this test are
removed.
Reviewers: Murali Basani <muralidhar.basani@aiven.io>, Chia-Ping Tsai
<chia7712@gmail.com>
When a broker becomes the leader of a `__transaction_state` partition,
`TransactionStateManager` loads the log up to the log end offset. A
newly elected leader can have records above the high watermark that are
not committed yet and can still be truncated if leadership moves again,
even with clean elections. The coordinator then acts on state that is
later rolled back. For example, it can send COMMIT markers for a
PREPARE_COMMIT that the next leader doesn't have, and that leader later
times out and aborts the transaction, so the transaction is committed on
some partitions and aborted on others.
Loading with `FetchIsolation.HIGH_WATERMARK` is not enough, because a
new leader's high watermark can lag behind the true one, so the load
would miss committed records.
The coordinator doesn't write to the partition while it is loading
(requests get `COORDINATOR_LOAD_IN_PROGRESS`), so the log end offset at
the start of the load doesn't change. Once the high watermark reaches
it, every record the load reads is committed. This change:
- In `loadTransactionsForTxnTopicPartition`, if the high watermark is
below the log end offset, registers a `PartitionListener` and schedules
the load once the high watermark reaches the log end offset. The high
watermark is checked again after the listener is added, since an update
between reading it and adding the listener isn't delivered to the
listener.
- Abandons the load (removes the `loadingPartitions` entry) if the
partition is not online, becomes a follower, fails or is deleted before
the high watermark catches up. This runs on the scheduler, since
listeners are invoked with partition locks held.
- Skips partitions that are loading in
`removeExpiredTransactionalIds()`. On a leader epoch bump, the previous
epoch's cache entry stays until the new load replaces it, and tombstones
written for it would move the log end offset while the load waits.
- Logs the time spent waiting for the high watermark as part of the load
time.
In the normal case the wait is about one follower fetch round trip. If
an ISR follower is unavailable, the load waits until it is removed from
the ISR (`replica.lag.time.max.ms`). This does not cover unclean leader
election, where committed records can be lost.
Testing: added unit tests to `TransactionStateManagerTest` for waiting
for the high watermark, abandoning on each partition transition and when
the partition is not online, the high watermark advancing during
listener registration, a leader epoch bump while waiting, and expiration
of a partition that is being reloaded.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Reviewers: Justine Olshan <jolshan@confluent.io>
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
A negative MaxRecords in a ShareFetch request (v1+) was not validated
and only surfaced as an empty response after the log fetch. Return
INVALID_REQUEST before the share session context is created. A value of
0 remains valid as clients send it for acknowledgement-only requests.
Reviewers: Andrew Schofield <aschofield@confluent.io>
Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
`ShareCoordinatorMetadataCacheHelperImplTest` covered
`isRequestHeaderV3Supported()` at `IBP_4_5_IV0` and when the metadata
version is unknown, but not at an older metadata version. This adds a
test that it returns false at `IBP_4_4_IV2`, the version just before the
request header v3 gate.
Reviewers: Andrew Schofield <aschofield@confluent.io>
`ScramMessages.ServerFirstMessage.toMessage()` builds the SCRAM
server-first-message with `String.format("r=%s,s=%s,i=%d", ...)`.
Without an explicit `Locale`, `%d` follows the JVM's default locale, and
on a broker whose default locale uses native digits (for example
`ar-EG`, `fa-IR`, `bn-BD`, `ne-NP`; JDK 9+ CLDR data) the iteration
count is emitted as `i=٤٠٩٦` instead of `i=4096`.
RFC 5802 requires ASCII digits, and `ServerFirstMessage` itself parses
the count with `i=(?<iterations>[0-9]+)`, so every SCRAM handshake
against such a broker fails with `SaslException: Invalid SCRAM server
first message format: r=...,i=٤٠٩٦` (non-Java clients reject it the same
way).
This passes `Locale.ROOT` to that `String.format` call, as already done
for `toLowerCase()` under KAFKA-20638 and for the
`tools`/`server-common` modules under KAFKA-20646 (which does not cover
`clients`), and adds a test that switches the default locale to `ar-EG`
(guarded with `@ResourceLock(Resources.LOCALE)`) and verifies the
message still uses ASCII digits. The other `String.format` calls in
`ScramMessages` only use `%s` and are not locale-sensitive.
Following review, it also removes
`ScramCredentialUtils.credentialToString()` / `credentialFromString()`:
the ZooKeeper-era credential string format had the same `%d` problem but
no callers left outside its own tests, so the methods, their constants
and the tests that only exercised them are gone.
`ScramCredentialUtils.createCache()` and its test are unchanged.
Reproduced against the released `kafka-clients` 4.4.0 jar on JDK 25:
```
java -Duser.language=ar -Duser.country=EG ...
server-first-message on the wire =
r=clientnonceservernonce,s=MDEy...,i=٤٠٩٦
client parse = FAILED: Invalid SCRAM server first
message format
```
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Reviewers: Andrew Schofield <aschofield@confluent.io>
---------
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
In `KRaftMetadataCache`, `getPartitionLeaderEndpoint` returns empty when
the previous leader is offline after the refactor.
With this PR, the method now returns `noNode` when the leader partition
exists, aligning with the original Scala implementation.
Changes are covered by unit tests. Please feel free to review.
Reviewers: Kevin Wu <kwu@confluent.io>, Justine Olshan
<jolshan@confluent.io>
Replace the for-loop with anyMatch in
maybeUpdateLimitOffsetsForStandbyChangelogs. Both short-circuit on the
first partition with buffered records, so there is no behavior change.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Migrate the Admin client timeout and close tests from
PlaintextAdminIntegrationTest to AdminClientTimeoutIntegrationTest. The
timeoutMs(0) tests now use assertTimeout to verify the option timeout is
honored, and tests that only need an invalid broker run with `@Test`
instead of `@ClusterTest`.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
During a long rebalance, a `StreamThread` can use native memory without
limit and fail with an out-of-memory error. When the uncommitted state
store bytes exceed `statestore.uncommitted.max.bytes`,
`StreamThread#maybeCommit` forces a commit. But `TaskManager` skips
commits during a rebalance, and `runOnceWithoutProcessingThreads`
continues to process records. Thus the uncommitted bytes continue to
increase. RocksDB keeps these bytes in a `WriteBatch` in native memory,
outside the block cache and the write buffers.
At the start of each iteration of the processing loop, `StreamThread`
now calls `maybePauseProcessingForUncommittedBytes()`, which checks
`shouldCommitDueToUncommittedBytes()`. Because `maybeCommit` commits at
the end of each iteration, the bytes exceed the limit at this point only
when that commit was skipped or failed. So the check needs no extra
state. While processing is paused, the thread does not process records
or run punctuators, because both write to the state stores. It continues
to poll, so the rebalance can complete, and it continues to call
`maybeCommit`. Processing resumes when a commit succeeds, or when a
commit on partition revocation flushes the buffers. The thread logs the
pause and the resume at INFO level.
This change decreases throughput during a long rebalance, but it
prevents the out-of-memory error. `GlobalStreamThread` does not need
this change, because one thread writes to its stores, checks the bytes
and flushes them, and nothing skips that flush. A separate pull request
will apply the same pause to processing threads.
Two new `StreamThreadTest` tests cover the change.
`shouldNotProcessOrPunctuateWhileUncommittedBytesExceedLimit` shows that
the thread polls and tries to commit, but does not process or punctuate,
while the commit is skipped.
`shouldResumeProcessingOnceUncommittedBytesAreCommitted` shows that
processing stays paused for the iteration in which the rebalance
completes, and resumes only after the commit succeeds.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Reviewers: Bill Bejeck <bbejeck@apache.org>
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Follow-up to #23165. Test-only change.
-
`testStreamsGroupEpochDoesNotIncreaseWhenEveryAssignmentConfigIsAtItsDefault`
now starts the group with
non-default assignment configs, so it fails if replaying a record
without configs does not reset them to the
defaults.
- Add
`testStreamsGroupEpochIncreasesWhenAnAssignmentConfigIsNotAtItsDefault`:
after the same replay, a group with
`num.standby.replicas=2` still bumps the group epoch, since the replayed
defaults differ from the current configs.
- Add `AssignmentConfigsImplTest` cases for `fromRecord` with an empty
list and with an unknown key.
- Use `AssignmentConfigsImpl.DEFAULT` instead of `new
AssignmentConfigsImpl(0, List.of())` in `GroupMetadataManagerTest`.
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
Widens the private subscribeInternal helpers to RebalanceListener and
retargets internal javadoc, including StickyAssignor's worked examples.
Log strings now say RebalanceListener.onPartitionsX; no assertions
depend on them.
Reviewers: TengYao Chi <frankvicky@apache.org>, Chia-Ping Tsai
<chia7712@gmail.com>
Migrates rather than suppressing: -Xfatal-warnings applies to
compileTestScala, and a @nowarn that suppresses nothing is itself fatal.
Reviewers: TengYao Chi <frankvicky@apache.org>
PlaintextConsumerAssignorsTest uses the ConsumerRebalanceListener when
it moved to Java in #23708. This was missed during #23108 port.
Migrate it, along with the MockConsumerTest, AsyncKafkaConsumerTest and
ConsumerSubscriptionStateTest cases where the old listener was
incidental.
Tests that cover the deprecated overloads or the 1-arg bridge stay on
the old API until it is removed. A dedicated test for the deprecated
subscribe overloads will be added in KAFKA-21365.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Follow-up to #23604. Several tests create a client, metrics, or log
manager that the cleanup method never closes, either because a local
variable shadows the field closed in `@AfterEach`, or because the field
set up in `@BeforeEach` is reassigned without closing the old instance.
This fixes the same pattern in `KafkaConsumerTest`, `MetricsTest`,
`LogManagerTest`, and `WorkerTaskTest`.
Generated-by: Claude Opus 5.5
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Follow-up to #23699 to apply the same changes to
ClusterInstance#deleteTopic.
ClusterInstance#deleteTopic did not wait for the DeleteTopics result, so
a failed deletion (e.g. the topic did not exist) was either ignored or
hidden by the waitTopicDeletion timeout. It now calls get() and wraps
the cause in a RuntimeException, so callers no longer need to handle
ExecutionException and InterruptedException.
- Use ClusterInstance#deleteTopic in tests that delete a topic and then
immediately wait for the deletion.
- Rename ClusterTestExtensionsTest#testVerifyTopicDeletion to
testDeleteTopic and verify that deleting a non-existent topic fails with
UnknownTopicOrPartitionException.
Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
Reviewing
ShareConsumerTest.testSubscriptionAndPollFollowedByTopicDeletion, it was
clear that it could be tidied up and redundant code removed.
Reviewers: Sushant Mahajan <sushant.mahajan88@gmail.com>
A share fetch hands its record budget to partitions in the order they
are read, so the partition read first takes the largest share. Two
things decided that order earlier and together they could starve a
partition, hence changed the behaviour.
Earlier:
- `SharePartitionManager` rotated the fetch list by `sessionEpoch %
size`. The client bumps the epoch on both
ShareFetch and ShareAcknowledge, so in a poll-and-commit loop fetch
epochs step by two and the rotation is
constant. One unpaired request flips the parity, and with two
partitions the rotation then undoes the
session reorder on every fetch, so the same partition is read first
until the other starves.
- `ShareSessionContext` moved served partitions to the end of the
session from inside the response iterator,
in the iteration order of the response map. That map had passed
through `HashMap` copies, so when every
partition returned records the session ended up in hash order and the
same partition led every fetch.
PR:
This change removes the epoch rotation from the fetch path and makes the
session reorder deterministic from the fetch order. The fetch order is
preserved through `ShareFetchUtils`, `ShareFetch` and `KafkaApis` with
`LinkedHashMap`s. `ShareSessionContext` reorders the session in one pass
before pruning the response: partitions without records keep their
position at the front, and partitions with records are appended in fetch
order rotated by one, so the partition read first is read last next
time. `PartitionRotateStrategy` is reduced to that rotation and the
unused `sessionEpoch` parameter of `fetchMessages` is dropped.
`ShareConsumerFairnessTest` adds cases for the epoch parity change, for
the first position rotating when all partitions return records with
explicit and implicit acknowledgement, and for an idle partition being
read first once it has records.
`SharePartitionManagerTest` verifies the session order across subsequent
fetches.
Reviewers: Abhinav Dixit <adixit@confluent.io>, Andrew Schofield
<aschofield@confluent.io>
Update MockConsumer Javadocs to reference `RebalanceListener`
The Javadocs for `MockConsumer#losePartitions` referenced
`ConsumerRebalanceListener`, while the callbacks are handled by
`RebalanceListener`. Update the links to point to the correct listener
type.
Reviewers: Andrew Schofield <aschofield@confluent.io>
Global stores share Streams' subtopology id counter, so the formatter
labels them with the lowest id not used by a subtopology instead of
starting at `subtopologies().size()`, which could collide.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>, TengYao Chi
<frankvicky@apache.org>
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Jira: https://issues.apache.org/jira/browse/KAFKA-20925
KIP-1313:
https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406624038/KIP-1313+Client+instance+ID+in+all+request+headers
This PR adds metadata version 4.5-IV0 and new versions of the share
group state RPCs that use the v3 request header: ReadShareGroupState v1,
WriteShareGroupState v2, InitializeShareGroupState v1,
DeleteShareGroupState v1 and ReadShareGroupStateSummary v2. The share
group state persister sends the new versions only when the metadata
version is 4.5-IV0 or later, so brokers keep using the v2 request header
until every broker in the cluster supports v3. Like the other v3 request
header versions, the new versions are marked unstable.
Reviewers: Andrew Schofield <aschofield@confluent.io>
This adds explicit tests for assignments containing a topic that a
member no longer subscribes to, covering both homogeneous and
heterogeneous subscriptions.
- testAssignWithUnsubscribedTopicHomogeneous: two members each hold one
partition of topic 1 and one partition of topic 2, after the group's
subscription changes to topic 1 only. It verifies that topic 2 is
revoked by
SimpleHomogeneousAssignmentBuilder#revokeUnassignablePartitions.
- testAssignWithUnsubscribedTopicHeterogeneous: member A drops topic 2
while member B still subscribes to it. It verifies that A's partition
of topic 2 is revoked and that B ends up with both partitions of
topic 2.
Topic 1 has four partitions, with only two initially assigned. Both
tests verify that each member retains its existing partition of topic 1
and receives one of the remaining partitions.
Reviewers: Andrew Schofield <aschofield@confluent.io>
The test tables in the JUnit job summary (Failed, Flaky, Skipped,
Quarantined and New) are printed in the order the reports are parsed,
which follows the filesystem order of `glob()` and is not stable across
environments. This sorts each table by class name and test name before
rendering.
This is a continuation of #21786, which was closed as stale.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
---------
Co-authored-by: Zhekai Chen <wrmxs227@gmail.com>
Mocking `Transformation.class` with Mockito returns a raw type, causing
an unchecked conversion to `Transformation<SourceRecord>`.
```
> Task :connect:runtime:compileTestJava
Note:
/Users/parker/opensource/kafka/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
uses unchecked or unsafe operations.
Note: Recompile with -Xlint:unchecked for details.
```
Add a local `@SuppressWarnings("unchecked")`, following the existing
approach in `TransformationStageTest.applyAndAssert`.
Validation:
- `./gradlew :connect:runtime:compileTestJava --rerun-tasks`
- The unchecked message for WorkerTest.java no longer appeared.
Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
ClusterInstance#createTopic and #createTopicWithAssignment did not wait
for the CreateTopics result, so a failed creation was hidden behind the
60s waitTopicCreation timeout, or ignored if the topic already existed.
They now call get() and wrap the cause in a RuntimeException.
InterruptedException is rethrown as InterruptException, so callers no
longer need to declare checked exceptions.
Also removes the now redundant throws clauses, assertDoesNotThrow
wrappers and pass-through helpers, and a duplicate createTopic call in
PlaintextConsumerTest#testSeek.
Follow-up to #23513.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Ken Huang
<s7133700@gmail.com>
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
The Hugo site build fails because docs/apis/internal-api-checker.md uses
`{{< param fullDotVersion >}}`, and `fullDotVersion` is not a defined
site param.
This PR replaces it with the hardcoded version `4.4.0`, matching how
other docs
pages (e.g. apis/_index.md, quickstart.md, docker.md) refer to the
release version.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
GroupSpec is part of the public assignor API, so custom assignors may
call isPartitionAssigned with a negative partition, or may have produced
a target assignment containing one. isPartitionAssigned threw an
IndexOutOfBoundsException in both cases. Report negative partitions as
unassigned, like other partitions which do not exist, and update the
tests.
Also update the javadoc of GroupSpecImpl's metadata image, and assign it
first in the constructor.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Reviewers: David Jacot <dajac@apache.org>
On the slow path, `DelayedShareFetch.combineLogReadResponse` reuses the
data read in `tryComplete` and issues a second `readFromLog` for
partitions that became acquirable only at completion. The second read
was issued even when the already fetched data covered `maxFetchRecords`,
in which case `SharePartition.acquire` receives a zero budget and
returns empty, so the read was wasted.
The remaining record budget is now computed from the already fetched
data and the second read is skipped when nothing remains. Skipped
partitions are absent from the response, as share sessions already do
for partitions without data, and their locks are released on the same
path as before.
Reviewers: Abhinav Dixit <adixit@confluent.io>, Andrew Schofield
<aschofield@confluent.io>
When the metadata layer (e.g. MetadataLoader) tries to read a slice of
records from KRaft which is greater than
KafkaRaftClient#MAX_BATCH_SIZE_BYTES, the RecordsIterator can return
records which are past the slice (HWM). As it iterates through this
slice of
records, FileRecords#readInto reads from disk the minimum of the bytes
until end of the file and the remaining space in the associated buffer.
In the case of the former, this can lead to a metadata error where the
node tries to snapshot with records past the HWM.
For the metadata error to occur, the following three pre-conditions must
be met:
1. One committed read handed to a listener is larger than 8 MiB.
2. The segment file has bytes after the HWM, meaning records this node
already appended but that aren't committed yet.
3. For the error to actually show up, the snapshot generator emits a
snapshot from the resulting image before the HWM catches up.
The underlying cause is the same as KAFKA-20841. Thanks to @asdf2014
for working on this issue as part of
https://github.com/apache/kafka/pull/22963.
Change `FileRecords#readInto` which reads from the underlying file to
the minimum of:
1. The remaining space in the associated buffer, which in the case of
RecordsIterator is the minimum of the FileRecords' record size and
the batch size defined by KRaft.
2. The end byte position of the FileRecords object, rather than the end
of the underlying file. This ensures when a caller constructs a
FileRecords slice of [start position, HWM), no bytes past the HWM are
read into memory.
Co-authored-by: Benedict Jin <asdf2014@apache.org>
Reviewers: José Armando García Sancio <jsancio@apache.org>, Jun Rao
<junrao@gmail.com>, Zhiyan Tang <ztang@confluent.io>, Chia-Ping Tsai
<chia7712@gmail.com>
Removing a test for KIP-557, which is a KIP that was completely disabled
due to data loss, with no plans to reenable
Reviewers: Bill Bejeck <bbejeck@apache.org>
DelayedShareFetch estimates the available bytes for minBytes by
comparing the end offset with the cached LogOffsetMetadata of the fetch
offset. The cache was kept on SharePartition, so it outlived the request
which populated it. A partition that is read but trimmed from the
response, or whose request fails, kept a valid cache entry although its
fetch offset did not advance, which requires resets and made the read
path depend on state left behind by earlier requests.
Keep the metadata inside DelayedShareFetch for the lifetime of the
request, as DelayedFetch does with FetchPartitionStatus. The retries of
tryComplete in purgatory still avoid reading from the log, and the cache
can no longer go stale across requests, so the remote fetch reset is not
needed.
Add ShareConsumerFairnessTest to verify that two partitions with a deep
backlog are both served under record_limit acquire mode.
There are also cases where if a record is released where it was earlier
fetch then request has to go through timeout to fetch data. Those
scenarios are also handled by this fix.
Reviewers: Andrew Schofield <aschofield@confluent.io>, Abhinav Dixit
<adixit@confluent.io>
---------
Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
Replace GroupSpecImpl's inverted target assignment with a BitSet of the
assigned partitions of each topic, computed from the members' current
assignments on the first call to isPartitionAssigned. Only the uniform
homogeneous assignor calls isPartitionAssigned, so other assignors no
longer compute it.
This makes build in TargetAssignmentBuilderBenchmark ~40% faster for
homogeneous groups, because isPartitionAssigned no longer looks up
partitions in the inverted target assignment's TimelineHashMaps, which
are slower than regular maps.
It also fixes a bug where the assignor saw the partitions of members
removed in the same request as still assigned. The bug is not reachable
today, since we do not compute a new target assignment when a member
leaves, only on the next heartbeat, and static member replacement keeps
the replaced member's partitions assigned.
Remove ModernGroup's inverted target assignment, since it is no longer
read. For share groups, it was never correct, because a partition can be
assigned to several members of a share group, while the inverted target
assignment holds a single member per partition.
Rename buildAssignment in TargetAssignmentBuilderBenchmark to
buildGroupSpecAndAssignment, which builds a new group spec on every
invocation, so that the cost of computing the assigned partitions is
included, and replace buildGroupSpecWithoutOffload and
buildGroupSpecWithOffload with an assignorOffload parameter.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Reviewers: David Jacot <dajac@apache.org>
This patch adds new exceptions and protocol errors required by the
Cluster Mirroring feature. No exception is marked as public for now. We
can change that later if needed.
Reviewers: Ken Huang <s7133700@gmail.com>, Luke Chen <showuon@gmail.com>
---------
Signed-off-by: Federico Valeri <fedevaleri@gmail.com>
The quotas of the uniform2 assignor tell how many partitions of every
topic each member gets. Once the steps have decided them, the assignor
places the partitions.
This patch adds `PartitionPlacement`, which places the partitions of a
topic on its subscribers, moving as few partitions as possible. Every
owner keeps its partitions with the lowest ids, up to its quota. The
other partitions, those without an owner and those beyond the quota of
their owner, go to the subscribers below their quota, each taking the
lowest ids left. A partition thus moves only when its owner owns more
partitions than its quota. `place` returns `false` when every partition
stays with its owner, so that the members keep their current sets of
partitions of the topic.
Reviewers: Sean Quah <squah@confluent.io>
- When record copy is enabled for a share group's DLQ,
ShareGroupDLQRecordHelper.headers() now preserves the original source
record's headers on the DLQ record instead of discarding them.
- If an original header's key collides with a DLQ context header that is
actually added for that record — unconditionally for
topic/partition/offset/group/delivery-count/cause. The original header
is dropped so the DLQ-computed value
is the only one present.
- Headers are now built RecordHeaders instead of hand-rolled array/size
bookkeeping, removing a class of size/counter-mismatch risk on future
edits.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Reviewers:
Andrew Schofield <aschofield@confluent.io>
Reviewers: Andrew Schofield <aschofield@confluent.io>
---------
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
The production OAUTHBEARER examples don't work as written. Add the
missing login callback handler to the broker and client examples, and
the org.apache.kafka.sasl.oauthbearer.allowed.urls system property,
which has defaulted to empty since 4.0.
Also add an integration test that runs the documented broker config with
a replicated topic.
Generated-by: Claude Opus 5.5
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
When a consumer group is reloaded, members using the classic protocol
are assigned the consumer protocol session timeout. This can cause
members to expire prematurely or remain in the group longer than their
configured timeout allows.
Update `GroupMetadataManager.onLoaded()` to check each member's protocol
and use the persisted `classicProtocolSessionTimeout` for classic
members. Members using the consumer protocol continue to use the
consumer protocol session timeout.
Reviewers: liuliu <liuliugit@gmail.com>, David Jacot <dajac@apache.org>
Follow-up to KAFKA-21289 (#23833), which fixed `release.py stage-docs`
for the Hugo-based kafka-site layout.
The site-docs tar carries the repo's top-level `LICENSE` and `NOTICE` at
its root. `stage-docs` extracts the tar with `tar xf ...
--strip-components 1` straight into `content/en/<version>/`, so those
two files get dumped into the kafka-site content tree alongside the
actual docs. They are not docs content and do not belong there — the
previous release slot (`content/en/43`) has neither, and the javadoc
tree already carries its own `legal/` license files under
`static/<version>/javadoc/`.
This was flagged on the 4.4 kafka-site PR, where `content/en/44/LICENSE`
and `content/en/44/NOTICE` showed up as spurious additions and had to be
removed by hand. Without this change they return on every `stage-docs`
run.
This patch drops `LICENSE` and `NOTICE` from the content directory after
extraction, mirroring the existing post-extraction cleanup that moves
`images/` out to `static/<version>/`.
No behavior change for any other staged file.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This PR moves DynamicLogConfig from Scala to Java, along with its
validation of local retention, remote copy lag and cordoned log dirs.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Wrap InitMapValue's partitions and
ShareGroupStatePartitionMetadataInfo's collections in unmodifiable
views, so that the share group state partition metadata cannot be
modified through them. Add withAddedPartitions and withRemovedPartitions
to InitMapValue to derive new values.
Use them in GroupMetadataManager and remove the copies that are no
longer needed. combineInitMaps reuses a value when only one of the maps
has the topic, and initializeShareGroupState no longer deep copies the
initializing topics to remove partitions in place.
initializedShareGroupPartitions and
sharePartitionsEligibleForOffsetDeletion no longer copy the partitions.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Reviewers: David Jacot <dajac@apache.org>
## Background
[KAFKA-20404](https://issues.apache.org/jira/browse/KAFKA-20404) tracks
replacing the bulk SpotBugs atomicity excludes (introduced when SpotBugs
was upgraded from 4.8.6 to 4.9.4) with per-field investigations. This PR
handles `DefaultStateUpdater$StateUpdaterThread`, the second class in
the series after `StreamThread` (#22320).
## Investigation
After removing `DefaultStateUpdater$StateUpdaterThread` from the two
bulk excludes it appeared in
(`AT_NONATOMIC_OPERATIONS_ON_SHARED_VARIABLE` at line 505 and
`AT_NONATOMIC_64BIT_PRIMITIVE` at line 544 on trunk), SpotBugs reports
exactly one field with warnings:
- `totalCheckpointLatency` at `DefaultStateUpdater.java:100`, flagged at
line 732 (`measureCheckpointLatency`) and line 750 (`recordMetrics`).
Walking the call sites:
- `measureCheckpointLatency` is `private` and only called from within
`StateUpdaterThread`: `restoreTasks`, `pauseTasks`, `resumeTasks`,
`removeUpdatingTask`, `maybeCheckpointTasks`.
- `recordMetrics` is `private` and only called from
`StateUpdaterThread.runOnce` (line 213).
- All callers run on the `StateUpdaterThread` itself (the inner class
extends `Thread`).
So `totalCheckpointLatency` is thread-confined and the SpotBugs warnings
are false positives.
## Changes
- Remove `DefaultStateUpdater$StateUpdaterThread` from the two bulk
`<Or>` blocks.
- Add two targeted `<Match>` blocks scoped to the class plus the
`totalCheckpointLatency` field, each with a rationale comment explaining
the thread-confinement, one per bug pattern.
## Verification
- `:streams:spotbugsMain --rerun-tasks` clean.
- `:streams:checkstyleMain` / `:streams:checkstyleTest` /
`:streams:spotbugsTest` clean.
- `:streams:test --tests "*DefaultStateUpdater*"` all green.
`GetTelemetrySubscriptions` v0 only treated `ZERO_UUID` or `null` as "no
client instance ID", so a request carrying `ONE_UUID` was accepted and
cached as a real client instance ID. Is is then rejected in
`PushTelemetry` by `isValidClientInstanceId` since it is a reserved
UUID.
So v0 now rejects reserved IDs with `INVALID_REQUEST`, matching
`PushTelemetry` and v1 `GetTelemetrySubscriptions`.
Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>
Split the benchmark into separate benchmarks for building the group
spec, the target assignment and the target assignment records. The group
spec is benchmarked both with and without assignor offload, because
offloading copies the group's members and target assignment so that the
assignor can run off the coordinator thread.
Populate a ConsumerGroup and read the members, target assignment and
inverted target assignment from it, instead of plain HashMaps. This
makes the benchmarks iterate over and copy the same TimelineHashMaps as
GroupMetadataManager, which matters most for the offload group spec
benchmark.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Reviewers: David Jacot <dajac@apache.org>
pick() keeps the tag keys in rack.aware.assignment.tags order, each only
if a process with room still meets it together with those kept so far.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Supersedes #21393.
Fixed typos in comments and updated ConfigDef exception message as
suggested in review.
Reviewers: Ken Huang <s7133700@gmail.com>
---------
Co-authored-by: Rbaddam <RBaddam@walmartlabs.com>
Ken Huang