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 2025/01/20 11:51:48 UTC

[PR] KAFKA-18604; Update transaction coordinator [kafka]

dajac opened a new pull request, #18636:
URL: https://github.com/apache/kafka/pull/18636

   This patch updates the transaction coordinator record to use the new coordinator record definition.
   
   ### 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-18604; Update transaction coordinator [kafka]

Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18636:
URL: https://github.com/apache/kafka/pull/18636#discussion_r1922713369


##########
transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionCoordinatorRecordSerde.java:
##########
@@ -16,32 +16,29 @@
  */
 package org.apache.kafka.coordinator.transaction;
 
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRecordSerde;
-import org.apache.kafka.coordinator.transaction.generated.TransactionLogKey;
-import org.apache.kafka.coordinator.transaction.generated.TransactionLogValue;
+import org.apache.kafka.coordinator.transaction.generated.CoordinatorRecordType;
 
 public class TransactionCoordinatorRecordSerde extends CoordinatorRecordSerde {
 
     @Override
     protected ApiMessage apiMessageKeyFor(short recordType) {
-        switch (recordType) {
-            case 0:
-                return new TransactionLogKey();
-            default:
-                throw new CoordinatorLoader.UnknownRecordTypeException(recordType);
+        try {
+            return CoordinatorRecordType.fromId(recordType).newRecordKey();
+        } catch (UnsupportedVersionException ex) {
+            throw new CoordinatorLoader.UnknownRecordTypeException(recordType);
         }
     }
 
     @Override
     protected ApiMessage apiMessageValueFor(short recordVersion) {

Review Comment:
   Again :). Fixed 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-18604; Update transaction coordinator [kafka]

Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #18636:
URL: https://github.com/apache/kafka/pull/18636


-- 
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-18604; Update transaction coordinator [kafka]

Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18636:
URL: https://github.com/apache/kafka/pull/18636#discussion_r1922494231


##########
transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionCoordinatorRecordSerde.java:
##########
@@ -16,32 +16,29 @@
  */
 package org.apache.kafka.coordinator.transaction;
 
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRecordSerde;
-import org.apache.kafka.coordinator.transaction.generated.TransactionLogKey;
-import org.apache.kafka.coordinator.transaction.generated.TransactionLogValue;
+import org.apache.kafka.coordinator.transaction.generated.CoordinatorRecordType;
 
 public class TransactionCoordinatorRecordSerde extends CoordinatorRecordSerde {
 
     @Override
     protected ApiMessage apiMessageKeyFor(short recordType) {
-        switch (recordType) {
-            case 0:
-                return new TransactionLogKey();
-            default:
-                throw new CoordinatorLoader.UnknownRecordTypeException(recordType);
+        try {
+            return CoordinatorRecordType.fromId(recordType).newRecordKey();
+        } catch (UnsupportedVersionException ex) {
+            throw new CoordinatorLoader.UnknownRecordTypeException(recordType);
         }
     }
 
     @Override
     protected ApiMessage apiMessageValueFor(short recordVersion) {

Review Comment:
   This really is the `recordType` again I think here. The version of the `TransactionLogValue` will have no bearing on the object which is created. It needs to be 0 here whether it's a v0 or v1 value.



-- 
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