Add PartitionsWithLateTransactionsCount to the Monitoring table with its
description, JMX name, and normal value.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Move the counters and timing state that only `consume` uses out of `run`
and into `consume`. This shortens the `consume` parameter list and drops
the redundant `joinStartMs` alias in favor of `startMs`. No behavior
change.
Reviewers: Uros (github:uros-b), Ken Huang <s7133700@gmail.com>,
Chia-Ping Tsai <chia7712@gmail.com>
As mentioned in
https://github.com/apache/kafka/pull/22017#discussion_r4429568025, the
class-level `@Timeout` on these test classes is redundant.
JUnit resolves a method-level `@Timeout` before the class-level one, and
a class-level `@Timeout` only applies to test methods that have no
timeout of their own.
`@ClusterTest`, `@ClusterTests` and `@ClusterTemplate` are
meta-annotated with `@Timeout(60)`, so each cluster test method
effectively carries a method-level timeout.
The changed test classes only contain cluster test methods, so I removed
the redundant `@Timeout` from them.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Add AdminClientTestUtils#currentLeader, which returns the current leader
of a partition without waiting. Remove AdminUtils#fetchOrWaitForLeader
and ClusterInstance#getLeaderBrokerId, and migrate all callers. Tests
that need a specific leader now wrap it in TestUtils#waitForCondition,
which also fixes the stale-leader race in LeaderElectionCommandTest.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Use `<>` for `StagedMergeIterator` constructor calls in the tests to
remove raw type warnings.
Reviewers: Yahya Kisana (github:ykisana), Chia-Ping Tsai
<chia7712@gmail.com>
`supports_consumer_close_timeout` compares against `DEV_VERSION`, which
is now `4.5.0-SNAPSHOT` after the 4.4 branch cut. That excludes 4.4.0
nodes even though the VerifiableConsumer `--close-timeout` option was
added in 4.4.0. Compare against `V_4_4_0` instead.
Follow-up to
https://github.com/apache/kafka/pull/23981#discussion_r4446688190.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This PR migrates `AllocateProducerIdsRequestTest` from the `core` module
to the `server` module and rewrites it in Java using `ClusterInstance`
and `@ClusterTest`.
Changes:
- Rewrite the two existing cases and find the active controller via
`controller().isActive()` instead of the broker's Raft client.
- Fix the typo in test names (`ProducersId` -> `ProducerIds`) and rename
`SentToNonController` to `SentToStandbyController`.
- Add tests for the remaining paths of
`handleAllocateProducerIdsRequest`: invalid broker epoch, unknown
broker, controller failover, and broker restart. The rejected cases also
verify that a rejected request does not consume a block.
- Remove the Scala test.
Test-only change, no behavior change.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
ByteBufferAccessor.readArray() and readByteBuffer() only checked that
the requested size did not exceed the bytes remaining, so a negative
size fell through to new byte[size] (NegativeArraySizeException) or
ByteBuffer.limit() (IllegalArgumentException) with no meaningful
message. A negative size can currently be produced while decoding wire
data: ByteUtils.readUnsignedVarint accepts 5-byte varints whose value
overflows into the sign bit and decodes as negative — see KAFKA-15423
for whether that behavior should change — and such a value is used as an
unknown tagged field's size (Readable.readUnknownTaggedField).
Reject negative sizes with the same descriptive RuntimeException used
for oversized reads. Added assertions to ByteBufferAccessorTest covering
negative sizes for readArray and readByteBuffer (in a dedicated
testReadByteBuffer test).
Reviewers: Sean Quah <squah@confluent.io>
Store the null indexes in ListDeserializer in a HashSet instead of an
ArrayList, so the per-entry null index check in the CONSTANT_SIZE
strategy does not scan the whole list. Also reject a record whose
declared number of entries is larger than the remaining bytes can hold
before reading the entries. Such a record would fail at the end of the
stream anyway.
Added a unit test in ListDeserializerTest.
Reviewers: Matthias J. Sax <matthias@confluent.io>
Compare the supplied password against a dummy value when the user is
unknown.
Added a unit test in PlainSaslServerTest.
Reviewers: Manikumar Reddy <manikumar@confluent.io>
Compare the supplied password with Utils.isEqualConstantTime instead of
String.equals, against a dummy value when the user is unknown.
Added unit tests in JaasBasicAuthFilterTest.
Reviewers: Manikumar Reddy <manikumar@confluent.io>
The `api-checker` module is a composite-included Gradle build, so its
tests and verification tasks are not run by the root `check` or `test`
tasks.
This change adds a dedicated `api-checker` CI job that:
- runs `check` for `api-checker:core`, `gradle-plugins`, and
`maven-plugin` on Java 25;
- uses the same Gradle cache and build scan settings as the `validate`
job;
- archives test and verification reports;
- is required by `CI checks completed`.
The job does not use the test catalog, since the included build does not
participate in Kafka's test-selection infrastructure.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Document in the `Admin#deleteRecords` javadoc that the returned future
fails with `OffsetOutOfRangeException` when the given offset is out of
range, and reference `RecordsToDelete#beforeOffset(long)` for the valid
values. Also clarify in `RecordsToDelete#beforeOffset(long)` that, apart
from `-1`, the offset must be at least 0 and at most the high watermark
of the partition.
Add `DeleteRecordsTest` to `clients-integration-tests` to cover three
cases: offset larger than the high watermark, offset smaller than `-1`,
and offset `-1` (truncate to the high watermark).
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Many Javadoc comments in the group coordinator module are out of date or
wrong. They name the wrong return type or record, describe share group
code with consumer group wording, miss `@param` tags, wrap `@link` in
double braces or contain typos. The exception messages in the consumer
and streams `CurrentAssignmentBuilder` also say "a unknown state".
This patch fixes them. It only changes Javadoc, comments and these two
exception messages.
Reviewers: Sean Quah <squah@confluent.io>, Chia-Ping Tsai
<chia7712@gmail.com>
Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
`PersisterStateManagerHandler.onCompleteCallback` has never been used
since it was added in KAFKA-17329. `setOnCompleteCallback` has no usage,
and `DefaultStatePersister` always passed `null` for the constructor
parameter.
This removes the field, the setter, the invocation in `onComplete`, and
the `Consumer<ClientResponse>` parameter from the handler constructors.
There is no behavior change.
Reviewers: liuliu <liuliugit@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
The generated message readers and the ArrayOf/CompactArrayOf parsers
reject any array that declares more than a fixed 1,000,000 elements, and
there is no way to raise it. Legitimate data can exceed that, for example
a producer snapshot that holds more than a million producer IDs.
Add a broker config, message.max.array.length (default 1,000,000, at
least 1), that controls this limit. It is read from the static server
config and applied once when the shared server starts, before the raft
client reads the metadata log, so the configured value also bounds the
metadata log and producer snapshots parsed during startup. The default
keeps the existing behaviour unchanged.
Co-authored-by: Truc Nguyen <119303279+trnguyencflt@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
DefaultForeignResponseSinkPartitioner routes foreign-key join responses
to the partition stored in the subscription record, without checking it
against the number of partitions. A negative value makes the
ProducerRecord constructor throw and fails the task, and a value that is
too large fails the send.
Fall back to the default partitioner, as for subscriptions without a
partition, when the partition is outside [0, numPartitions), and log a
warning.
Added unit tests in KTableImplTest and
KTableKTableForeignKeyJoinScenarioTest.
Reviewers: Matthias J. Sax <matthias@confluent.io>, liuliu
<liuliugit@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
Add 4.4.0 to the system tests: register `LATEST_4_4` in `version.py` and
include it in the compatibility, upgrade, transactions and streams test
matrices; fetch the 4.4.0 binaries and streams test jar in the Docker
and Vagrant setup; add the `upgrade-system-tests-44` module and
`UPGRADE_FROM_44`.
Reviewers: Maros Orsak <maros.orsak159@gmail.com>, Lianet Magrans
<lmagrans@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>
Validate the topic and partition offset counts in
PositionSerde.deserialize against the remaining bytes in the buffer
before creating the maps. Negative or too large counts throw an
IllegalArgumentException.
Added unit tests in PositionSerdeTest.
Reviewers: Matthias J. Sax <matthias@confluent.io>
Handle an empty or null value of the changelog version and position
headers in ChangelogRecordDeserializationHelper. Such records are
skipped with a warning, the same way an unknown version is handled. A
missing position header still throws a StreamsException.
Added unit tests in ChangelogRecordDeserializationHelperTest.
Reviewers: Matthias J. Sax <matthias@confluent.io>
Use Byte.BYTES, Integer.BYTES and Long.BYTES instead of Byte.SIZE,
Integer.SIZE and Long.SIZE when sizing the buffer in
PositionSerde.serialize, so the serialized position has no trailing zero
bytes. deserialize ignores trailing bytes, so existing positions can
still be read.
Added unit tests in PositionSerdeTest.
Reviewers: Matthias J. Sax <matthias@confluent.io>
Several methods in group-coordinator implement an interface method or
override a superclass method without the `@Override` annotation.
This patch adds the annotation to `ClassicGroup`, `ModernGroup`,
`ConsumerGroup`, `ShareGroup`, `StreamsGroup`, `TasksTuple`,
`ShareGroupAutoOffsetResetStrategy.Validator` and `GroupConfigManager`.
Reviewers: Sean Quah <squah@confluent.io>, Yahya Kisana
(github:ykisana), Chia-Ping Tsai <chia7712@gmail.com>
Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
Migrate `ClientQuotaMetadataManager` from Scala in the `core` module to
Java in the `metadata` module, consolidating its logic into
`DynamicClientQuotaPublisher`.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
`startup` and `shutdown` in `GroupCoordinatorService` call
`isActive.set(...)` with the value that the preceding `compareAndSet`
has just stored, and `buildErrorResponse` for AlterShareGroupOffsets
sets the responses on the result and then overwrites them.
This patch removes the redundant `isActive.set` calls and the first
`setResponses` call.
The group coordinator logs use several prefixes for the group id
(`[GroupId {}]`, `[GroupId={}]`, `[Group {}]`), put the group id inside
the sentence, or omit it, which makes the logs of one group hard to
filter.
This patch prefixes every log about a single group in
`GroupMetadataManager`, `OffsetMetadataManager`, `ClassicGroup` and
`StreamsGroupTopologyDescriptionManager` with `[GroupId {}]`, with the
group id as the first argument, and rephrases the messages that named
the group in the sentence. The logs that are not about a single group,
such as the transaction completion logs and the topology description
cleanup cycle logs, are unchanged.
Reviewers: Sean Quah <squah@confluent.io>
The "Computed a new target assignment" logs for share and streams groups
pass the assignor object and print its `toString`. Only the built-in
streams assignors override it to return their name, so the share log and
the streams log of a custom assignor print the default
`Object.toString`, while the consumer group log prints the assignor
name.
This patch passes `name()` of the assignor in the share and streams
logs, so all three group types log the assignor name.
Reviewers: Sean Quah <squah@confluent.io>
The `thenCompose` functions in `shareGroupHeartbeat`, `deleteShareGroup`
and `persisterDeleteToGroupIdErrorMap` always return an already
completed future.
This patch uses `thenApply` in these three methods, which avoids
allocating that future.
Reviewers: Sean Quah <squah@confluent.io>
The `metadataImage` parameter of `maybeUpdateStreamsTargetAssignment`
shadows the field of the same name, and both callers pass that field.
This patch removes the parameter, so the method uses the field directly.
The following response builders only produced a v0-shaped response, so
RequestResponseTest never exercised the fields added in later versions.
This PR makes them accept `version` and populate version-specific
fields:
- VoteResponse: NodeEndpoints (v1+)
- BeginQuorumEpochResponse: NodeEndpoints (v1+)
- EndQuorumEpochResponse: NodeEndpoints (v1+)
- DescribeQuorumResponse: LastFetchTimestamp and LastCaughtUpTimestamp
(v1+);
ReplicaDirectoryId, ErrorMessage (top-level and partition-level) and
Nodes (v2+)
EndQuorumEpochResponse also sets LeaderId so it matches the
NodeEndpoint.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Follow-up to review comments on #23821.
The share group state RPCs capped their latest version with a hardcoded
`LATEST_VERSION_BEFORE_REQUEST_HEADER_V3` constant per request, chosen
by a boolean from `MetadataVersion`. `MetadataVersion` now returns the
request header version directly, and `ApiKeys` looks up the latest API
version that uses that header version or lower, so the builders take the
header version and the per-request constants are removed.
Reviewers: Andrew Schofield <aschofield@confluent.io>
The fields of `ModernGroupMember` are not `final`, although only its
constructor assigns them and its subclasses `ConsumerGroupMember` and
`ShareGroupMember` are documented as immutable.
This patch makes all nine fields `protected final`.
Reviewers: Sean Quah <squah@confluent.io>
The Javadoc of `Group` names a nonexistent `offsets.retention.ms`
config, describes `stateAsString` as returning the group type, says
"genetic groups", uses `{{@link ...}}` double braces and omits some
`@param` tags.
This patch fixes them.
Reviewers: Sean Quah <squah@confluent.io>
The `toString` methods of `ClassicGroup`, `ResolvedRegularExpression`,
`SubscribedTopicDescriberImpl` and `MemberAssignmentImpl` print
`ClassicGroupMetadata`, `RegularExpressionMetadata`,
`SubscribedTopicMetadata` and `MemberAssignment`, which are not their
class names.
This patch makes each one print its own class name.
Reviewers: Sean Quah <squah@confluent.io>, Murali Basani
<muralidhar.basani@aiven.io>
The `earliest()` and `latest()` factories of
`ShareGroupAutoOffsetResetStrategy` have no callers, since the code uses
the `EARLIEST` and `LATEST` constants.
This patch removes both methods.
Reviewers: Sean Quah <squah@confluent.io>, Chia-Ping Tsai
<chia7712@gmail.com>
`OverlayMap` calls `containsKey` followed by `get` with the same key in
`get`, `put`, `remove` and the entry iterator, which looks the key up
twice.
This patch replaces each pair with a single `get` followed by a null
check. This is equivalent because neither the base map nor the updates
may hold null values. The class Javadoc now states this explicitly.
Reviewers: Sean Quah <squah@confluent.io>
The Javadoc of `GroupCoordinator` documents the heartbeat requests as
responses, misspells `TxnOffsetCommitRequest`, says "tight" instead of
"tied" for the buffer supplier, and has no `@param requireStable` on
`fetchOffsets`.
This patch fixes these, along with the type names of the `listGroups`
and `deleteShareGroupOffsets` requests, the missing blank lines in
`alterShareGroupOffsets` and `groupConfig`, and the missing final
periods in some `@param` descriptions.
Reviewers: Sean Quah <squah@confluent.io>, Chia-Ping Tsai
<chia7712@gmail.com>
Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
Fix the UUIDSerializer Javadoc to use `serializer.encoding` property
names instead of `deserializer.encoding`, matching the implementation.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Several log and exception messages in `GroupMetadataManager` have a
stray `}` in a placeholder, a missing space, a doubled period or the
wrong term, and one share group tombstone check does not use the message
of its consumer and streams counterparts.
This patch fixes them.
Reviewers: Sean Quah <squah@confluent.io>
The `TopicMetadata` record in the `streams` package is no longer used by
main code, which relies on `CoordinatorMetadataImage.TopicMetadata`.
Only its own test still references it.
This patch removes the record and its test.
Reviewers: Sean Quah <squah@confluent.io>, Lucas Brutschy
<lbrutschy@confluent.io>
The `toString` of `ShareGroupMember` appends a stray quote after
`memberEpoch`, `previousMemberEpoch` and `subscribedTopicNames`, which
are not quoted fields.
This patch removes those three quotes.
Reviewers: Sean Quah <squah@confluent.io>
The `group.consumer.assignors` doc says "customer assignors" instead of
"custom assignors", and the `offsets.load.buffer.size` doc renders a
double space where its two string parts are joined.
This patch fixes both strings.
Reviewers: Sean Quah <squah@confluent.io>
`GroupCoordinatorService` checks `request.topology() != null` right
after `throwIfNull` on the same value, so the check is always true.
This patch removes the check and calls `throwIfInvalidTopology`
directly.
Reviewers: Sean Quah <squah@confluent.io>, Lucas Brutschy
<lbrutschy@confluent.io>
`addPendingSyncMember` and `removePendingSyncMember` throw when the
member is not in the group, but their messages say it is already a
stable member, and the remove variant says it attempts to add.
This patch makes both messages name the operation and say that the
member is not in the group.
Reviewers: Sean Quah <squah@confluent.io>
A new StreamsGroupHeartbeatResponse status code is a semantic contract
change: it requires a KIP and a request-version bump, and the broker
must gate the new code so it is only sent to clients that advertise the
introducing version (as it does today for MISSING_CLIENT_TAGS, withheld
from v0 clients). If a broker bug sends a code an older client does not
recognize, the client decodes every status via Status.fromCode, which
throws -- and that throw, raised while handling the heartbeat response,
is currently swallowed (KAFKA-20860): the client keeps heartbeating,
advances its member epoch, and silently ignores the assignment and every
one after it, while the coordinator sees a healthy member.
This PR rejects the response as a whole, before any of it is applied, if
it carries a status this client cannot interpret: log it, surface a
KafkaException through the uncaught-exception handler, and fail the
member, so the error is visible instead of silently dropping every
assignment.
Also complete a pending leave when a fatal error arrives on the leave
heartbeat.
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
SubscriptionSendProcessorSupplier took the primary partition from the
record metadata, which is -1 for records forwarded from a punctuator.
The response is then sent to partition -1 and fails the task. Use the
partition of the task instead, which is the partition of the left-table
record in all cases.
Reviewers: Bill Bejeck <bbejeck@apache.org>
Co-authored-by: Matthias J. Sax <matthias@ibm.com>
Check the record length in SubscriptionWrapperSerde.deserialize against
the fixed header (version and instruction bytes, the optional 16-byte
hash and the 4-byte primary partition for version 1) before reading it,
and reject unknown versions before parsing. A truncated record now
throws a SerializationException instead of a NegativeArraySizeException
or BufferUnderflowException.
Added unit tests in SubscriptionWrapperSerdeTest.
Reviewers: Matthias J. Sax <matthias@confluent.io>
Kouki
YANG-SYUAN CHOU
Parker Chang
Murali Basani
Russole Chen
Aaron Thomas
Aashish Kumar
winfriedgerlach
Lucy Liu
David JacotandClaude Sonnet 5.5
Yen-Ting Li
Omnia Ibrahim
majialong
Alan Lau
Matthias J. Sax
Andrew Schofield