You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2025/05/30 11:00:22 UTC
[PR] KAFKA-19338: Error on read/write of uninitialized share part. [kafka]
smjn opened a new pull request, #19861:
URL: https://github.com/apache/kafka/pull/19861
- Currently, read and write share state requests were allowed on uninitialized share partitions (share partitions on which initializeState has NOT been called). This should not be allowed.
- This PR addresses the concern by adding error checks on read and write. Other requests are allowed (initialize, readSummary, alter).
- Refactored `ShareCoordinatorShardTest` to reduce redundancy and added some new tests.
- Some request/response classes have also been reformatted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19338: Error on read/write of uninitialized share part. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19861:
URL: https://github.com/apache/kafka/pull/19861#discussion_r2122975078
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -658,28 +645,14 @@ private CoordinatorRecord generateShareStateRecord(
) {
long timestamp = time.milliseconds();
int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
- if (!shareStateMap.containsKey(key)) {
- // Since this is the first time we are getting a write request for key, we should be creating a share snapshot record.
- // The incoming partition data could have overlapping state batches, we must merge them
- return ShareCoordinatorRecordHelpers.newShareSnapshotRecord(
- key.groupId(), key.topicId(), partitionData.partition(),
- new ShareGroupOffset.Builder()
- .setSnapshotEpoch(0)
- .setStartOffset(partitionData.startOffset())
- .setLeaderEpoch(partitionData.leaderEpoch())
- .setStateEpoch(partitionData.stateEpoch())
- .setStateBatches(mergeBatches(List.of(), partitionData))
- .setCreateTimestamp(timestamp)
- .setWriteTimestamp(timestamp)
- .build());
- } else if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
+ if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
Review Comment:
Hi @AndrewJSchofield ,
Thanks for the review.
No, it is not possible since this method is called from `ShareCoordinatorShard.writeState` and `ShareCoordinatorShard.readStateAndMaybeUpdateLeaderEpoch`. But before calling it - both methods do error checking (`maybeGetWriteStateError`, `maybeGetReadStateError`) where they explicitly check that the `shareStateMap` is not empty and return "read/write on uninitialized share partition" error.
New tests `testWriteFailsOnUninitializedPartition` and `testReadFailsOnUninitializedPartition` add proof for the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19338: Error on read/write of uninitialized share part. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19861:
URL: https://github.com/apache/kafka/pull/19861#discussion_r2122975078
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -658,28 +645,14 @@ private CoordinatorRecord generateShareStateRecord(
) {
long timestamp = time.milliseconds();
int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
- if (!shareStateMap.containsKey(key)) {
- // Since this is the first time we are getting a write request for key, we should be creating a share snapshot record.
- // The incoming partition data could have overlapping state batches, we must merge them
- return ShareCoordinatorRecordHelpers.newShareSnapshotRecord(
- key.groupId(), key.topicId(), partitionData.partition(),
- new ShareGroupOffset.Builder()
- .setSnapshotEpoch(0)
- .setStartOffset(partitionData.startOffset())
- .setLeaderEpoch(partitionData.leaderEpoch())
- .setStateEpoch(partitionData.stateEpoch())
- .setStateBatches(mergeBatches(List.of(), partitionData))
- .setCreateTimestamp(timestamp)
- .setWriteTimestamp(timestamp)
- .build());
- } else if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
+ if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
Review Comment:
Hi @AndrewJSchofield ,
Thanks for the review.
No, it is not possible since this method is called from `ShareCoordinatorShard.writeState` and `ShareCoordinatorShard.readStateAndMaybeUpdateLeaderEpoch`. But before calling it - both methods do error checking (`maybeGetWriteStateError`, `maybeGetReadStateError`) where they explicitly check that the `shareStateMap` is not empty and return "read/write on uninitialized share partition" error.
New tests `testWriteFailsOnUninitializedPartition` and `testReadFailsOnUninitializedPartition` add proof for the same.
Will add additional javadoc to clarify.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19338: Error on read/write of uninitialized share part. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19861:
URL: https://github.com/apache/kafka/pull/19861#discussion_r2121064720
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -644,8 +632,7 @@ public CoordinatorResult<Void, CoordinatorRecord> maybeCleanupShareState(Set<Uui
/**
* Util method to generate a ShareSnapshot or ShareUpdate type record for a key, based on various conditions.
* <p>
- * If no snapshot has been created for the key => create a new ShareSnapshot record
- * else if number of ShareUpdate records for key >= max allowed per snapshot per key => create a new ShareSnapshot record
+ * Ff number of ShareUpdate records for key >= max allowed per snapshot per key or stateEpoch is highest seen => create a new ShareSnapshot record
Review Comment:
nit: What's "Ff"?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -658,28 +645,14 @@ private CoordinatorRecord generateShareStateRecord(
) {
long timestamp = time.milliseconds();
int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
- if (!shareStateMap.containsKey(key)) {
- // Since this is the first time we are getting a write request for key, we should be creating a share snapshot record.
- // The incoming partition data could have overlapping state batches, we must merge them
- return ShareCoordinatorRecordHelpers.newShareSnapshotRecord(
- key.groupId(), key.topicId(), partitionData.partition(),
- new ShareGroupOffset.Builder()
- .setSnapshotEpoch(0)
- .setStartOffset(partitionData.startOffset())
- .setLeaderEpoch(partitionData.leaderEpoch())
- .setStateEpoch(partitionData.stateEpoch())
- .setStateBatches(mergeBatches(List.of(), partitionData))
- .setCreateTimestamp(timestamp)
- .setWriteTimestamp(timestamp)
- .build());
- } else if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
+ if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
Review Comment:
I believe it's possible for `shareStateMap` not to contain the key.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19338: Error on read/write of uninitialized share part. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #19861:
URL: https://github.com/apache/kafka/pull/19861#discussion_r2122975078
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -658,28 +645,14 @@ private CoordinatorRecord generateShareStateRecord(
) {
long timestamp = time.milliseconds();
int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
- if (!shareStateMap.containsKey(key)) {
- // Since this is the first time we are getting a write request for key, we should be creating a share snapshot record.
- // The incoming partition data could have overlapping state batches, we must merge them
- return ShareCoordinatorRecordHelpers.newShareSnapshotRecord(
- key.groupId(), key.topicId(), partitionData.partition(),
- new ShareGroupOffset.Builder()
- .setSnapshotEpoch(0)
- .setStartOffset(partitionData.startOffset())
- .setLeaderEpoch(partitionData.leaderEpoch())
- .setStateEpoch(partitionData.stateEpoch())
- .setStateBatches(mergeBatches(List.of(), partitionData))
- .setCreateTimestamp(timestamp)
- .setWriteTimestamp(timestamp)
- .build());
- } else if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
+ if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
Review Comment:
Hi @AndrewJSchofield ,
Thanks for the review.
No, it is not possible since this method is called from `ShareCoordinatorShard.writeState` and `ShareCoordinatorShard.readStateAndMaybeUpdateLeaderEpoch`. But before calling it - both methods do error checking (`maybeGetWriteStateError`, `maybeGetReadStateError`) where they are explicitly check that the `shareStateMap` is not empty and return "read/write on uninitialized share partition" error.
New tests `testWriteFailsOnUninitializedPartition` and `testReadFailsOnUninitializedPartition` add proof for the same.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -658,28 +645,14 @@ private CoordinatorRecord generateShareStateRecord(
) {
long timestamp = time.milliseconds();
int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
- if (!shareStateMap.containsKey(key)) {
- // Since this is the first time we are getting a write request for key, we should be creating a share snapshot record.
- // The incoming partition data could have overlapping state batches, we must merge them
- return ShareCoordinatorRecordHelpers.newShareSnapshotRecord(
- key.groupId(), key.topicId(), partitionData.partition(),
- new ShareGroupOffset.Builder()
- .setSnapshotEpoch(0)
- .setStartOffset(partitionData.startOffset())
- .setLeaderEpoch(partitionData.leaderEpoch())
- .setStateEpoch(partitionData.stateEpoch())
- .setStateBatches(mergeBatches(List.of(), partitionData))
- .setCreateTimestamp(timestamp)
- .setWriteTimestamp(timestamp)
- .build());
- } else if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
+ if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit || partitionData.stateEpoch() > shareStateMap.get(key).stateEpoch()) {
Review Comment:
Hi @AndrewJSchofield ,
Thanks for the review.
No, it is not possible since this method is called from `ShareCoordinatorShard.writeState` and `ShareCoordinatorShard.readStateAndMaybeUpdateLeaderEpoch`. But before calling it - both methods do error checking (`maybeGetWriteStateError`, `maybeGetReadStateError`) where they explicitly check that the `shareStateMap` is not empty and return "read/write on uninitialized share partition" error.
New tests `testWriteFailsOnUninitializedPartition` and `testReadFailsOnUninitializedPartition` add proof for the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19338: Error on read/write of uninitialized share part. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #19861:
URL: https://github.com/apache/kafka/pull/19861
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org