Repository: apache/kafka
File: transaction-coordinator/src/test/java/org/apache/kafka/coordinator/transaction/TransactionLogTest.java (Java)

Code before:
        transactionLogValue.set("transaction_partitions", new Struct[]{txnPartitions});
        transactionLogValue.set("transaction_last_update_timestamp_ms", 2000L);
        transactionLogValue.set("transaction_start_timestamp_ms", 3000L);
        var txnLogValueTaggedFields = new TreeMap<Integer, Object>();
        txnLogValueTaggedFields.put(100, "foo");
        txnLogValueTaggedFields.put(101, 4000);
        transactionLogValue.set("_tagged_fields", txnLogValueTaggedFields);

        // Prepare the buffer.
        var buffer = allocate(Short.BYTES + transactionLogValue.sizeOf());
        buffer.putShort((short) 1); // Add 1 as version.
        transactionLogValue.writeTo(buffer);
        buffer.flip();

        // Read the buffer with the real schema and verify that tagged
        // fields were read but ignored.
        buffer.getShort(); // Skip version.
        var value = new TransactionLogValue(new ByteBufferAccessor(buffer), (short) 1);
        assertEquals(List.of(100, 101), value.unknownTaggedFields().stream().map(RawTaggedField::tag).toList());
        assertEquals(List.of(100, 101), value.transactionPartitions().get(0).unknownTaggedFields().stream().map(RawTaggedField::tag).toList());

        // Read the buffer with readTxnRecordValue.
        buffer.rewind();
        var metadata = TransactionLog.readTxnRecordValue("transaction-id", buffer);

        assertFalse(metadata == null, "Expected transaction metadata but got none");
        assertEquals(1000L, metadata.producerId());
        assertEquals(100, metadata.producerEpoch());
        assertEquals(1000, metadata.txnTimeoutMs());
        assertEquals(TransactionState.COMPLETE_COMMIT, metadata.state());
        assertEquals(Set.of(new TopicPartition("topic", 1)), metadata.topicPartitions());
        assertEquals(2000L, metadata.txnLastUpdateTimestamp());
        assertEquals(3000L, metadata.txnStartTimestamp());
    }

    @Test
    void testReadTxnRecordKeyThrowsOnUnknownVersion() {
        var unknownRecord = MessageUtil.toVersionPrefixedBytes(Short.MAX_VALUE, new TransactionLogKey());
        var exception = assertThrows(IllegalStateException.class,
            () -> TransactionLog.readTxnRecordKey(wrap(unknownRecord)));
        assertTrue(exception.getMessage().contains("Unknown version " + Short.MAX_VALUE));
    }
   
    @Test
    void shouldReturnNullForTombstoneRecord() {
        assertEquals(null, TransactionLog.readTxnRecordValue("transaction-id", null));
    }
}

Review comment (suggestion):
```java
assertNotNull(metadata, "Expected transaction metadata but got none");
```

Code after:
        transactionLogValue.set("transaction_status", TransactionState.COMPLETE_COMMIT.id());
        transactionLogValue.set("transaction_partitions", new Struct[]{txnPartitions});
        transactionLogValue.set("transaction_last_update_timestamp_ms", 2000L);
        transactionLogValue.set("transaction_start_timestamp_ms", 3000L);
        var txnLogValueTaggedFields = new TreeMap<Integer, Object>();
        txnLogValueTaggedFields.put(100, "foo");
        txnLogValueTaggedFields.put(101, 4000);
        transactionLogValue.set("_tagged_fields", txnLogValueTaggedFields);

        // Prepare the buffer.
        var buffer = allocate(Short.BYTES + transactionLogValue.sizeOf());
        buffer.putShort((short) 1); // Add 1 as version.
        transactionLogValue.writeTo(buffer);
        buffer.flip();

        // Read the buffer with the real schema and verify that tagged
        // fields were read but ignored.
        buffer.getShort(); // Skip version.
        var value = new TransactionLogValue(new ByteBufferAccessor(buffer), (short) 1);
        assertEquals(List.of(100, 101), value.unknownTaggedFields().stream().map(RawTaggedField::tag).toList());
        assertEquals(List.of(100, 101), value.transactionPartitions().get(0).unknownTaggedFields().stream().map(RawTaggedField::tag).toList());

        // Read the buffer with readTxnRecordValue.
        buffer.rewind();
        var metadata = TransactionLog.readTxnRecordValue("transaction-id", buffer);

        assertNotNull(metadata, "Expected transaction metadata but got none");
        assertEquals(1000L, metadata.producerId());
        assertEquals(100, metadata.producerEpoch());
        assertEquals(1000, metadata.txnTimeoutMs());
        assertEquals(TransactionState.COMPLETE_COMMIT, metadata.state());
        assertEquals(Set.of(new TopicPartition("topic", 1)), metadata.topicPartitions());
        assertEquals(2000L, metadata.txnLastUpdateTimestamp());
        assertEquals(3000L, metadata.txnStartTimestamp());
    }

    @Test
    void testReadTxnRecordKeyThrowsOnUnknownVersion() {
        var unknownRecord = MessageUtil.toVersionPrefixedBytes(Short.MAX_VALUE, new TransactionLogKey());
        var exception = assertThrows(IllegalStateException.class,
            () -> TransactionLog.readTxnRecordKey(wrap(unknownRecord)));
        assertTrue(exception.getMessage().contains("Unknown version " + Short.MAX_VALUE));
    }
   
    @Test
    void shouldReturnNullForTombstoneRecord() {
        assertNull(TransactionLog.readTxnRecordValue("transaction-id", null));
    }
}