You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dajac (via GitHub)" <gi...@apache.org> on 2024/01/12 16:53:04 UTC
[PR] KAFKA-14505-7; [7/7] Always materialize the most recent committed offset [kafka]
dajac opened a new pull request, #15183:
URL: https://github.com/apache/kafka/pull/15183
When transactional offset commits are eventually committed, we must always keep the most recent committed when we have a mix of transactional and regular offset commits. We achieve this by storing the offset of the offset commit record along side the committed offset in memory. Without preserving information of the commit record offset, compaction of the __consumer_offsets topic itself may result in the wrong offset commit being materialized.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #15183:
URL: https://github.com/apache/kafka/pull/15183
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15183:
URL: https://github.com/apache/kafka/pull/15183#issuecomment-1905440595
> Unrelated to the PR -- do we know why `Build / JDK 21 and Scala 2.13 / testDescribeQuorumReplicationSuccessful [1] Type=Raft-Combined, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest` and some other tests are reported 3 times on the test output? I didn't think we ran tests 3 times.
```
Build / JDK 21 and Scala 2.13 / testDescribeQuorumReplicationSuccessful [1] Type=Raft-Combined, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest
1m 7s
Build / JDK 21 and Scala 2.13 / testDescribeQuorumReplicationSuccessful [2] Type=Raft-Isolated, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest
1m 8s
Build / JDK 21 and Scala 2.13 / testDescribeQuorumReplicationSuccessful [5] Type=Raft-Combined, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest
```
There are not exactly the same. This is a parameterized test and I suppose that not all the arguments are printed out.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1459300270
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetAndMetadata.java:
##########
@@ -92,30 +117,34 @@ public boolean equals(Object o) {
OffsetAndMetadata that = (OffsetAndMetadata) o;
- if (offset != that.offset) return false;
+ if (committedOffset != that.committedOffset) return false;
if (commitTimestampMs != that.commitTimestampMs) return false;
- if (!leaderEpoch.equals(that.leaderEpoch)) return false;
- if (!metadata.equals(that.metadata)) return false;
- return expireTimestampMs.equals(that.expireTimestampMs);
+ if (recordOffset != that.recordOffset) return false;
+ if (!Objects.equals(leaderEpoch, that.leaderEpoch)) return false;
Review Comment:
huh. Guess I never used it.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1458501802
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetAndMetadata.java:
##########
@@ -92,30 +117,34 @@ public boolean equals(Object o) {
OffsetAndMetadata that = (OffsetAndMetadata) o;
- if (offset != that.offset) return false;
+ if (committedOffset != that.committedOffset) return false;
if (commitTimestampMs != that.commitTimestampMs) return false;
- if (!leaderEpoch.equals(that.leaderEpoch)) return false;
- if (!metadata.equals(that.metadata)) return false;
- return expireTimestampMs.equals(that.expireTimestampMs);
+ if (recordOffset != that.recordOffset) return false;
+ if (!Objects.equals(leaderEpoch, that.leaderEpoch)) return false;
Review Comment:
I don't think so. There is a feature to auto-generate equals and hashCode. It has been around for a while now.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15183:
URL: https://github.com/apache/kafka/pull/15183#issuecomment-1900735439
> we could delete all the pending transactional offsets when a new offset record is replayed.
Hmmm does this work for LSO correctly? I know that the consumer offset topic has slightly different logic, but I think we would still need to keep pending offsets until the transaction is committed?
Maybe I need to step back and see how last stable offset works now.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1457831330
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -918,7 +920,7 @@ public void replay(
groupId,
topic,
partition,
- OffsetAndMetadata.fromRecord(value)
+ OffsetAndMetadata.fromRecord(offset, value)
Review Comment:
I wonder if that is easy to get confused. And if there is a way to make it easier to known what the parameters mean
--
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-14505; [7/7] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1454285529
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -979,15 +981,30 @@ public void replayEndTransactionMarker(
pendingOffsets.offsetsByGroup.forEach((groupId, topicOffsets) -> {
topicOffsets.forEach((topicName, partitionOffsets) -> {
partitionOffsets.forEach((partitionId, offsetAndMetadata) -> {
- log.debug("Committed transaction offset commit for producer id {} in group {} " +
- "with topic {}, partition {}, and offset {}.",
- producerId, groupId, topicName, partitionId, offsetAndMetadata);
- offsets.put(
+ OffsetAndMetadata existingOffsetAndMetadata = offsets.get(
groupId,
topicName,
- partitionId,
- offsetAndMetadata
+ partitionId
);
+
+ // We always keep the most recent committed offset when we have a mix of transactional and regular
+ // offset commits. Without preserving information of the commit record offset, compaction of the
+ // __consumer_offsets topic itself may result in the wrong offset commit being materialized.
+ if (existingOffsetAndMetadata == null || offsetAndMetadata.recordOffset > existingOffsetAndMetadata.recordOffset) {
+ log.debug("Committed transactional offset commit {} for producer id {} in group {} " +
+ "with topic {} and partition {}.",
+ offsetAndMetadata, producerId, groupId, topicName, partitionId);
+ offsets.put(
Review Comment:
if the offset is not the latest offset were we incorrectly saying the last committed offset was the latest one?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -979,15 +981,30 @@ public void replayEndTransactionMarker(
pendingOffsets.offsetsByGroup.forEach((groupId, topicOffsets) -> {
topicOffsets.forEach((topicName, partitionOffsets) -> {
partitionOffsets.forEach((partitionId, offsetAndMetadata) -> {
- log.debug("Committed transaction offset commit for producer id {} in group {} " +
- "with topic {}, partition {}, and offset {}.",
- producerId, groupId, topicName, partitionId, offsetAndMetadata);
- offsets.put(
+ OffsetAndMetadata existingOffsetAndMetadata = offsets.get(
groupId,
topicName,
- partitionId,
- offsetAndMetadata
+ partitionId
);
+
+ // We always keep the most recent committed offset when we have a mix of transactional and regular
+ // offset commits. Without preserving information of the commit record offset, compaction of the
+ // __consumer_offsets topic itself may result in the wrong offset commit being materialized.
+ if (existingOffsetAndMetadata == null || offsetAndMetadata.recordOffset > existingOffsetAndMetadata.recordOffset) {
+ log.debug("Committed transactional offset commit {} for producer id {} in group {} " +
+ "with topic {} and partition {}.",
+ offsetAndMetadata, producerId, groupId, topicName, partitionId);
+ offsets.put(
Review Comment:
if the offset is not the latest offset were we incorrectly saying the last transactionally committed offset was the latest one?
--
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-14505; [7/7] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1454285529
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -979,15 +981,30 @@ public void replayEndTransactionMarker(
pendingOffsets.offsetsByGroup.forEach((groupId, topicOffsets) -> {
topicOffsets.forEach((topicName, partitionOffsets) -> {
partitionOffsets.forEach((partitionId, offsetAndMetadata) -> {
- log.debug("Committed transaction offset commit for producer id {} in group {} " +
- "with topic {}, partition {}, and offset {}.",
- producerId, groupId, topicName, partitionId, offsetAndMetadata);
- offsets.put(
+ OffsetAndMetadata existingOffsetAndMetadata = offsets.get(
groupId,
topicName,
- partitionId,
- offsetAndMetadata
+ partitionId
);
+
+ // We always keep the most recent committed offset when we have a mix of transactional and regular
+ // offset commits. Without preserving information of the commit record offset, compaction of the
+ // __consumer_offsets topic itself may result in the wrong offset commit being materialized.
+ if (existingOffsetAndMetadata == null || offsetAndMetadata.recordOffset > existingOffsetAndMetadata.recordOffset) {
+ log.debug("Committed transactional offset commit {} for producer id {} in group {} " +
+ "with topic {} and partition {}.",
+ offsetAndMetadata, producerId, groupId, topicName, partitionId);
+ offsets.put(
Review Comment:
if the offset is not the latest offset were we incorrectly saying the last transactional commit offset was the latest one?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -979,15 +981,30 @@ public void replayEndTransactionMarker(
pendingOffsets.offsetsByGroup.forEach((groupId, topicOffsets) -> {
topicOffsets.forEach((topicName, partitionOffsets) -> {
partitionOffsets.forEach((partitionId, offsetAndMetadata) -> {
- log.debug("Committed transaction offset commit for producer id {} in group {} " +
- "with topic {}, partition {}, and offset {}.",
- producerId, groupId, topicName, partitionId, offsetAndMetadata);
- offsets.put(
+ OffsetAndMetadata existingOffsetAndMetadata = offsets.get(
groupId,
topicName,
- partitionId,
- offsetAndMetadata
+ partitionId
);
+
+ // We always keep the most recent committed offset when we have a mix of transactional and regular
+ // offset commits. Without preserving information of the commit record offset, compaction of the
+ // __consumer_offsets topic itself may result in the wrong offset commit being materialized.
+ if (existingOffsetAndMetadata == null || offsetAndMetadata.recordOffset > existingOffsetAndMetadata.recordOffset) {
+ log.debug("Committed transactional offset commit {} for producer id {} in group {} " +
+ "with topic {} and partition {}.",
+ offsetAndMetadata, producerId, groupId, topicName, partitionId);
+ offsets.put(
Review Comment:
if the offset is not the latest offset were we incorrectly saying the transactional commit offset was the latest one?
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1458504559
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -918,7 +920,7 @@ public void replay(
groupId,
topic,
partition,
- OffsetAndMetadata.fromRecord(value)
+ OffsetAndMetadata.fromRecord(offset, value)
Review Comment:
I agree that it is a bit confusing. Let me rename `offset` to `recordOffset` to make it clearer.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1457470738
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetAndMetadata.java:
##########
@@ -92,30 +117,34 @@ public boolean equals(Object o) {
OffsetAndMetadata that = (OffsetAndMetadata) o;
- if (offset != that.offset) return false;
+ if (committedOffset != that.committedOffset) return false;
if (commitTimestampMs != that.commitTimestampMs) return false;
- if (!leaderEpoch.equals(that.leaderEpoch)) return false;
- if (!metadata.equals(that.metadata)) return false;
- return expireTimestampMs.equals(that.expireTimestampMs);
+ if (recordOffset != that.recordOffset) return false;
+ if (!Objects.equals(leaderEpoch, that.leaderEpoch)) return false;
Review Comment:
IntelliJ auto-generated this one. We should ask it. :). I suppose that it uses Objects.equals for the reason you mentioned.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15183:
URL: https://github.com/apache/kafka/pull/15183#issuecomment-1904351570
> I don’t see the relation with the LSO here. The LSO is tracked in the log layer so we don’t influence it here at all. In the group coordinator, the only requirement is to ensure that the last committed offset stored in the log is materialized in memory.
Sorry I'm probably missing something, but why do we need the last committed offset if it is not related to LSO.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15183:
URL: https://github.com/apache/kafka/pull/15183#issuecomment-1904840972
Unrelated to the PR -- do we know why
`Build / JDK 21 and Scala 2.13 / testDescribeQuorumReplicationSuccessful [1] Type=Raft-Combined, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest` and some other tests are reported 3 times on the test output? I didn't think we ran tests 3 times.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1458508459
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java:
##########
@@ -712,13 +712,14 @@ public void run() {
try {
// Apply the records to the state machine.
if (result.replayRecords()) {
- result.records().forEach(record ->
+ for (int i = 0; i < result.records().size(); i++) {
context.coordinator.replay(
+ prevLastWrittenOffset + i,
Review Comment:
Sure.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15183:
URL: https://github.com/apache/kafka/pull/15183#issuecomment-1899919345
For the reference, I was thinking about an alternative approach. Instead of keeping track of the offset of the record in the log for every committed offsets, we could delete all the pending transactional offsets when a new offset record is replayed. The downside of this approach is that we would need to restructure how pending transactional offsets are stored in order to have an efficient way to find all the pending offsets for a given group, topic and partition tuple. Otherwise, the replay would become less efficient than it is today.
--
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-14505; [7/7] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1454282345
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -918,7 +920,7 @@ public void replay(
groupId,
topic,
partition,
- OffsetAndMetadata.fromRecord(value)
+ OffsetAndMetadata.fromRecord(offset, value)
Review Comment:
it is interesting that the offset passed in here is new (probably not the offset that OffsetAndMetadata is referring to)
--
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-14505; [7/7] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1454293498
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java:
##########
@@ -712,13 +712,14 @@ public void run() {
try {
// Apply the records to the state machine.
if (result.replayRecords()) {
- result.records().forEach(record ->
+ for (int i = 0; i < result.records().size(); i++) {
context.coordinator.replay(
+ prevLastWrittenOffset + i,
Review Comment:
is this the only way to get the offsets? 😅
--
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-14505; [7/7] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1454280167
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetAndMetadata.java:
##########
@@ -92,30 +117,34 @@ public boolean equals(Object o) {
OffsetAndMetadata that = (OffsetAndMetadata) o;
- if (offset != that.offset) return false;
+ if (committedOffset != that.committedOffset) return false;
if (commitTimestampMs != that.commitTimestampMs) return false;
- if (!leaderEpoch.equals(that.leaderEpoch)) return false;
- if (!metadata.equals(that.metadata)) return false;
- return expireTimestampMs.equals(that.expireTimestampMs);
+ if (recordOffset != that.recordOffset) return false;
+ if (!Objects.equals(leaderEpoch, that.leaderEpoch)) return false;
Review Comment:
any reason why we changed from the .equals to Objects.equals? I guess it is safer if one of them can be null
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1457475930
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java:
##########
@@ -712,13 +712,14 @@ public void run() {
try {
// Apply the records to the state machine.
if (result.replayRecords()) {
- result.records().forEach(record ->
+ for (int i = 0; i < result.records().size(); i++) {
context.coordinator.replay(
+ prevLastWrittenOffset + i,
Review Comment:
yes. the only way to speculate about what the offset will be in the log based on the last written offset. as the coordinator is the single writer to the log, it works.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1457471555
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -918,7 +920,7 @@ public void replay(
groupId,
topic,
partition,
- OffsetAndMetadata.fromRecord(value)
+ OffsetAndMetadata.fromRecord(offset, value)
Review Comment:
yeah, this offset is the offset of the record, it is not the committed offset in the record.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1458064289
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetAndMetadata.java:
##########
@@ -92,30 +117,34 @@ public boolean equals(Object o) {
OffsetAndMetadata that = (OffsetAndMetadata) o;
- if (offset != that.offset) return false;
+ if (committedOffset != that.committedOffset) return false;
if (commitTimestampMs != that.commitTimestampMs) return false;
- if (!leaderEpoch.equals(that.leaderEpoch)) return false;
- if (!metadata.equals(that.metadata)) return false;
- return expireTimestampMs.equals(that.expireTimestampMs);
+ if (recordOffset != that.recordOffset) return false;
+ if (!Objects.equals(leaderEpoch, that.leaderEpoch)) return false;
Review Comment:
I see. Is this using AI 👀
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1457832081
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java:
##########
@@ -712,13 +712,14 @@ public void run() {
try {
// Apply the records to the state machine.
if (result.replayRecords()) {
- result.records().forEach(record ->
+ for (int i = 0; i < result.records().size(); i++) {
context.coordinator.replay(
+ prevLastWrittenOffset + i,
Review Comment:
could we maybe leave a comment about that?
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1457474626
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -979,15 +981,30 @@ public void replayEndTransactionMarker(
pendingOffsets.offsetsByGroup.forEach((groupId, topicOffsets) -> {
topicOffsets.forEach((topicName, partitionOffsets) -> {
partitionOffsets.forEach((partitionId, offsetAndMetadata) -> {
- log.debug("Committed transaction offset commit for producer id {} in group {} " +
- "with topic {}, partition {}, and offset {}.",
- producerId, groupId, topicName, partitionId, offsetAndMetadata);
- offsets.put(
+ OffsetAndMetadata existingOffsetAndMetadata = offsets.get(
groupId,
topicName,
- partitionId,
- offsetAndMetadata
+ partitionId
);
+
+ // We always keep the most recent committed offset when we have a mix of transactional and regular
+ // offset commits. Without preserving information of the commit record offset, compaction of the
+ // __consumer_offsets topic itself may result in the wrong offset commit being materialized.
+ if (existingOffsetAndMetadata == null || offsetAndMetadata.recordOffset > existingOffsetAndMetadata.recordOffset) {
+ log.debug("Committed transactional offset commit {} for producer id {} in group {} " +
+ "with topic {} and partition {}.",
+ offsetAndMetadata, producerId, groupId, topicName, partitionId);
+ offsets.put(
Review Comment:
yeah, that's right. the overall idea is to ensure that the last committed offset in the log is the one meterialized in memory. imagine the following sequence:
1. committed transactional offset 100 as part of transaction X. it goes to the pending structure.
2. committed regular offset 101. it goes to the main structure.
3. transaction X is committed. putting 100 in the main structure is wrong here because the last one in the log is 101.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15183:
URL: https://github.com/apache/kafka/pull/15183#issuecomment-1902066144
I don’t see the relation with the LSO here. The LSO is tracked in the log layer so we don’t influence it here at all. In the group coordinator, the only requirement is to ensure that the last committed offset stored in the log is materialized in memory.
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1462074651
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -2691,6 +2716,57 @@ public void testReplayTransactionEndMarkerWithCommit() {
));
}
+ @Test
+ public void testReplayTransactionEndMarkerKeepsTheMostRecentCommittedOffset() {
+ OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
+
+ // Add pending transactional offset commit for producer id 5.
+ verifyTransactionalReplay(context, 5L, "foo", "bar", 0, new OffsetAndMetadata(
+ 0L,
+ 100L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Add regular offset commit.
+ verifyReplay(context, "foo", "bar", 0, new OffsetAndMetadata(
+ 1L,
+ 101L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Replaying an end marker to commit transaction of producer id 5.
+ context.replayEndTransactionMarker(5L, TransactionResult.COMMIT);
+
+ // The pending offset is removed...
+ assertNull(context.offsetMetadataManager.pendingTransactionalOffset(
+ 5L,
+ "foo",
+ "bar",
+ 0
+ ));
Review Comment:
should we add a check that the pending txn offset exists before replaying the end txn marker in L2744?
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1462083032
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -2691,6 +2716,57 @@ public void testReplayTransactionEndMarkerWithCommit() {
));
}
+ @Test
+ public void testReplayTransactionEndMarkerKeepsTheMostRecentCommittedOffset() {
+ OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
+
+ // Add pending transactional offset commit for producer id 5.
+ verifyTransactionalReplay(context, 5L, "foo", "bar", 0, new OffsetAndMetadata(
+ 0L,
+ 100L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Add regular offset commit.
+ verifyReplay(context, "foo", "bar", 0, new OffsetAndMetadata(
+ 1L,
+ 101L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Replaying an end marker to commit transaction of producer id 5.
+ context.replayEndTransactionMarker(5L, TransactionResult.COMMIT);
+
+ // The pending offset is removed...
+ assertNull(context.offsetMetadataManager.pendingTransactionalOffset(
+ 5L,
+ "foo",
+ "bar",
+ 0
+ ));
Review Comment:
`verifyTransactionalReplay` replays and verifies. Is it what you want?
--
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-14505; [7/N] Always materialize the most recent committed offset [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1462298053
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -2691,6 +2716,57 @@ public void testReplayTransactionEndMarkerWithCommit() {
));
}
+ @Test
+ public void testReplayTransactionEndMarkerKeepsTheMostRecentCommittedOffset() {
+ OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
+
+ // Add pending transactional offset commit for producer id 5.
+ verifyTransactionalReplay(context, 5L, "foo", "bar", 0, new OffsetAndMetadata(
+ 0L,
+ 100L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Add regular offset commit.
+ verifyReplay(context, "foo", "bar", 0, new OffsetAndMetadata(
+ 1L,
+ 101L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Replaying an end marker to commit transaction of producer id 5.
+ context.replayEndTransactionMarker(5L, TransactionResult.COMMIT);
+
+ // The pending offset is removed...
+ assertNull(context.offsetMetadataManager.pendingTransactionalOffset(
+ 5L,
+ "foo",
+ "bar",
+ 0
+ ));
Review Comment:
oh, got it. yeah, i see we already verify. thanks
--
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