Skip to content

MINOR: Update GroupCoordinator interface to use AuthorizableRequestContext instead of RequestContext #19485

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Apr 16, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ public static String maybeTruncateReason(final String reason) {
*
* @return whether a known member id is required or not.
*/
public static boolean requiresKnownMemberId(short apiVersion) {
public static boolean requiresKnownMemberId(int apiVersion) {
return apiVersion >= 4;
}

Expand All @@ -117,7 +117,7 @@ public static boolean requiresKnownMemberId(short apiVersion) {
*/
public static boolean requiresKnownMemberId(
JoinGroupRequestData request,
short apiVersion
int apiVersion
) {
return request.groupInstanceId() == null
&& request.memberId().equals(UNKNOWN_MEMBER_ID)
Expand Down Expand Up @@ -150,7 +150,7 @@ public static boolean requiresKnownMemberId(
* @return whether the version supports skipping assignment.
*/

public static boolean supportsSkippingAssignment(short apiVersion) {
public static boolean supportsSkippingAssignment(int apiVersion) {
return apiVersion >= 9;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2236,7 +2236,7 @@ public <T> CompletableFuture<T> scheduleTransactionalWriteOperation(
short producerEpoch,
Duration timeout,
CoordinatorWriteOperation<S, T, U> op,
Short apiVersion
int apiVersion
) {
throwIfNotRunning();
log.debug("Scheduled execution of transactional write operation {}.", name);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ CompletableFuture<VerificationGuard> maybeStartTransactionVerification(
String transactionalId,
long producerId,
short producerEpoch,
short apiVersion
int apiVersion
) throws KafkaException;

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ public CompletableFuture<VerificationGuard> maybeStartTransactionVerification(
String transactionalId,
long producerId,
short producerEpoch,
short apiVersion
int apiVersion
) throws KafkaException {
return CompletableFuture.completedFuture(new VerificationGuard());
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ class CoordinatorPartitionWriter(
transactionalId: String,
producerId: Long,
producerEpoch: Short,
apiVersion: Short
apiVersion: Int
): CompletableFuture[VerificationGuard] = {
val transactionSupportedOperation = AddPartitionsToTxnManager.txnOffsetCommitRequestVersionToTransactionSupportedOperation(apiVersion)
val future = new CompletableFuture[VerificationGuard]()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ object AddPartitionsToTxnManager {
}
}

def txnOffsetCommitRequestVersionToTransactionSupportedOperation(version: Short): TransactionSupportedOperation = {
def txnOffsetCommitRequestVersionToTransactionSupportedOperation(version: Int): TransactionSupportedOperation = {
if (version > 4) {
addPartition
} else if (version > 3) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ class CoordinatorPartitionWriterTest {
"transactional-id",
10L,
5.toShort,
ApiKeys.TXN_OFFSET_COMMIT.latestVersion()
ApiKeys.TXN_OFFSET_COMMIT.latestVersion().toInt
)

if (error == Errors.NONE) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ void validateOffsetCommit(
String groupInstanceId,
int generationIdOrMemberEpoch,
boolean isTransactional,
short apiVersion
int apiVersion

) throws KafkaException;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,12 +49,12 @@
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.BufferSupplier;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;

import java.time.Duration;
import java.util.List;
Expand All @@ -80,7 +80,7 @@ public interface GroupCoordinator {
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<ConsumerGroupHeartbeatResponseData> consumerGroupHeartbeat(
RequestContext context,
AuthorizableRequestContext context,
ConsumerGroupHeartbeatRequestData request
);

Expand All @@ -94,7 +94,7 @@ CompletableFuture<ConsumerGroupHeartbeatResponseData> consumerGroupHeartbeat(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<StreamsGroupHeartbeatResult> streamsGroupHeartbeat(
RequestContext context,
AuthorizableRequestContext context,
StreamsGroupHeartbeatRequestData request
);

Expand All @@ -108,7 +108,7 @@ CompletableFuture<StreamsGroupHeartbeatResult> streamsGroupHeartbeat(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<ShareGroupHeartbeatResponseData> shareGroupHeartbeat(
RequestContext context,
AuthorizableRequestContext context,
ShareGroupHeartbeatRequestData request
);

Expand All @@ -123,7 +123,7 @@ CompletableFuture<ShareGroupHeartbeatResponseData> shareGroupHeartbeat(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<JoinGroupResponseData> joinGroup(
RequestContext context,
AuthorizableRequestContext context,
JoinGroupRequestData request,
BufferSupplier bufferSupplier
);
Expand All @@ -139,7 +139,7 @@ CompletableFuture<JoinGroupResponseData> joinGroup(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<SyncGroupResponseData> syncGroup(
RequestContext context,
AuthorizableRequestContext context,
SyncGroupRequestData request,
BufferSupplier bufferSupplier
);
Expand All @@ -154,7 +154,7 @@ CompletableFuture<SyncGroupResponseData> syncGroup(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<HeartbeatResponseData> heartbeat(
RequestContext context,
AuthorizableRequestContext context,
HeartbeatRequestData request
);

Expand All @@ -168,7 +168,7 @@ CompletableFuture<HeartbeatResponseData> heartbeat(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<LeaveGroupResponseData> leaveGroup(
RequestContext context,
AuthorizableRequestContext context,
LeaveGroupRequestData request
);

Expand All @@ -182,7 +182,7 @@ CompletableFuture<LeaveGroupResponseData> leaveGroup(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<ListGroupsResponseData> listGroups(
RequestContext context,
AuthorizableRequestContext context,
ListGroupsRequestData request
);

Expand All @@ -196,7 +196,7 @@ CompletableFuture<ListGroupsResponseData> listGroups(
* The error codes of the results are set to indicate the errors occurred during the execution.
*/
CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> describeGroups(
RequestContext context,
AuthorizableRequestContext context,
List<String> groupIds
);

Expand All @@ -209,7 +209,7 @@ CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> describeGroup
* @return A future yielding the results or an exception.
*/
CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> consumerGroupDescribe(
RequestContext context,
AuthorizableRequestContext context,
List<String> groupIds
);

Expand All @@ -222,7 +222,7 @@ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> consum
* @return A future yielding the results or an exception.
*/
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> streamsGroupDescribe(
RequestContext context,
AuthorizableRequestContext context,
List<String> groupIds
);

Expand All @@ -235,7 +235,7 @@ CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> streams
* @return A future yielding the results or an exception.
*/
CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> shareGroupDescribe(
RequestContext context,
AuthorizableRequestContext context,
List<String> groupIds
);

Expand All @@ -250,7 +250,7 @@ CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> shareGrou
* The error codes of the results are set to indicate the errors occurred during the execution.
*/
CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> deleteGroups(
RequestContext context,
AuthorizableRequestContext context,
List<String> groupIds,
BufferSupplier bufferSupplier
);
Expand All @@ -265,7 +265,7 @@ CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> delet
* The error codes of the results are set to indicate the errors occurred during the execution.
*/
CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchOffsets(
RequestContext context,
AuthorizableRequestContext context,
OffsetFetchRequestData.OffsetFetchRequestGroup request,
boolean requireStable
);
Expand All @@ -280,7 +280,7 @@ CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchOffsets
* The error codes of the results are set to indicate the errors occurred during the execution.
*/
CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchAllOffsets(
RequestContext context,
AuthorizableRequestContext context,
OffsetFetchRequestData.OffsetFetchRequestGroup request,
boolean requireStable
);
Expand All @@ -295,7 +295,7 @@ CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchAllOffs
* The error codes of the response are set to indicate the errors occurred during the execution.
*/
CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> describeShareGroupOffsets(
RequestContext context,
AuthorizableRequestContext context,
DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup request
);

Expand All @@ -309,7 +309,7 @@ CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffset
* The error codes of the response are set to indicate the errors occurred during the execution.
*/
CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> describeShareGroupAllOffsets(
RequestContext context,
AuthorizableRequestContext context,
DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup request
);

Expand All @@ -323,7 +323,7 @@ CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffset
* The error codes of the response are set to indicate the errors occurred during the execution.
*/
CompletableFuture<DeleteShareGroupOffsetsResponseData> deleteShareGroupOffsets(
RequestContext context,
AuthorizableRequestContext context,
DeleteShareGroupOffsetsRequestData request
);

Expand All @@ -338,7 +338,7 @@ CompletableFuture<DeleteShareGroupOffsetsResponseData> deleteShareGroupOffsets(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<OffsetCommitResponseData> commitOffsets(
RequestContext context,
AuthorizableRequestContext context,
OffsetCommitRequestData request,
BufferSupplier bufferSupplier
);
Expand All @@ -354,7 +354,7 @@ CompletableFuture<OffsetCommitResponseData> commitOffsets(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<TxnOffsetCommitResponseData> commitTransactionalOffsets(
RequestContext context,
AuthorizableRequestContext context,
TxnOffsetCommitRequestData request,
BufferSupplier bufferSupplier
);
Expand All @@ -370,7 +370,7 @@ CompletableFuture<TxnOffsetCommitResponseData> commitTransactionalOffsets(
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
CompletableFuture<OffsetDeleteResponseData> deleteOffsets(
RequestContext context,
AuthorizableRequestContext context,
OffsetDeleteRequestData request,
BufferSupplier bufferSupplier
);
Expand Down
Loading