You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gongxuanzhang (via GitHub)" <gi...@apache.org> on 2025/02/28 01:24:28 UTC
[PR] KAFKA-18882:Move BaseKey, TxnKey, and UnknownKey to transaction-coordinator module [kafka]
gongxuanzhang opened a new pull request, #19054:
URL: https://github.com/apache/kafka/pull/19054
Move BaseKey, TxnKey, and UnknownKey to transaction-coordinator module
--
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-18882:Move BaseKey, TxnKey, and UnknownKey to transaction-coordinator module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19054:
URL: https://github.com/apache/kafka/pull/19054#discussion_r1978709881
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala:
##########
@@ -235,7 +237,9 @@ class TransactionLogTest {
def testReadTxnRecordKeyCanReadUnknownMessage(): Unit = {
val record = new TransactionLogKey()
val unknownRecord = MessageUtil.toVersionPrefixedBytes(Short.MaxValue, record)
- val key = TransactionLog.readTxnRecordKey(ByteBuffer.wrap(unknownRecord))
- assertEquals(UnknownKey(Short.MaxValue), key)
+ TransactionLog.readTxnRecordKey(ByteBuffer.wrap(unknownRecord))match {
Review Comment:
`) match {`
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -465,25 +465,21 @@ class TransactionStateManager(brokerId: Int,
fileRecords.readInto(buffer, 0)
MemoryRecords.readableRecords(buffer)
}
-
memRecords.batches.forEach { batch =>
for (record <- batch.asScala) {
require(record.hasKey, "Transaction state log's key should not be null")
TransactionLog.readTxnRecordKey(record.key) match {
- case txnKey: TxnKey =>
- // load transaction metadata along with transaction state
Review Comment:
please keep this comment
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala:
##########
@@ -90,19 +90,15 @@ object TransactionLog {
/**
* Decodes the transaction log messages' key
*
- * @return the key
+ * @return left with the version if the key is not a transaction log key, right with the transactional id otherwise
*/
- def readTxnRecordKey(buffer: ByteBuffer): BaseKey = {
+ def readTxnRecordKey(buffer: ByteBuffer): Either[Short,String] = {
Review Comment:
`Either[Short, String]`
--
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-18882:Move BaseKey, TxnKey, and UnknownKey to transaction-coordinator module [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #19054:
URL: https://github.com/apache/kafka/pull/19054#discussion_r1975528720
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala:
##########
@@ -96,12 +97,9 @@ object TransactionLog {
val version = buffer.getShort
if (version == CoordinatorRecordType.TRANSACTION_LOG.id) {
val value = new TransactionLogKey(new ByteBufferAccessor(buffer), 0.toShort)
- TxnKey(
- version = version,
- transactionalId = value.transactionalId
- )
+ new TxnKey(version,value.transactionalId)
Review Comment:
nit: space
`new TxnKey(version, value.transactionalId)`
--
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-18882:Move BaseKey, TxnKey, and UnknownKey to transaction-coordinator module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19054:
URL: https://github.com/apache/kafka/pull/19054#issuecomment-2692901838
After reading the code again, those case classes are used internally. Maybe we can refactor `readTxnRecordKey` to return `Either[Short, String]` instead?
--
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-18882:Move BaseKey, TxnKey, and UnknownKey to transaction-coordinator module [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #19054:
URL: https://github.com/apache/kafka/pull/19054#discussion_r1976585693
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala:
##########
@@ -96,12 +97,9 @@ object TransactionLog {
val version = buffer.getShort
if (version == CoordinatorRecordType.TRANSACTION_LOG.id) {
val value = new TransactionLogKey(new ByteBufferAccessor(buffer), 0.toShort)
- TxnKey(
- version = version,
- transactionalId = value.transactionalId
- )
+ new TxnKey(version,value.transactionalId)
Review Comment:
it has been modified
--
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-18882:Move BaseKey, TxnKey, and UnknownKey to transaction-coordinator module [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19054:
URL: https://github.com/apache/kafka/pull/19054#discussion_r1979382072
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -889,10 +889,13 @@ class TransactionStateManagerTest {
appendedRecords.values.foreach { batches =>
batches.foreach { records =>
records.records.forEach { record =>
- val transactionalId = TransactionLog.readTxnRecordKey(record.key).transactionalId
- assertNull(record.value)
- expiredTransactionalIds += transactionalId
- assertEquals(Right(None), transactionManager.getTransactionState(transactionalId))
+ TransactionLog.readTxnRecordKey(record.key) match {
+ case Right(transactionalId) =>
+ assertNull(record.value)
+ expiredTransactionalIds += transactionalId
+ assertEquals(Right(None), transactionManager.getTransactionState(transactionalId))
+ case Left(value) => fail("Failed to read transactional id from tombstone: " + value)
Review Comment:
nit: `fail(s"Failed to read transactional id from tombstone: $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
Re: [PR] KAFKA-18882 Remove BaseKey, TxnKey, and UnknownKey [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #19054:
URL: https://github.com/apache/kafka/pull/19054
--
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