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

Code before:
    public static byte[] keyToBytes(String transactionalId) {
        return MessageUtil.toCoordinatorTypePrefixedBytes(
                new TransactionLogKey().setTransactionalId(transactionalId)
        );
    }

    /**
     * Generates the payload bytes for transaction log message value
     *
     * @return value payload bytes
     */
    public static byte[] valueToBytes(TxnTransitMetadata txnMetadata,
                                      TransactionVersion transactionVersionLevel) {
        if (txnMetadata.txnState() == TransactionState.EMPTY && !txnMetadata.topicPartitions().isEmpty()) {
            throw new IllegalStateException("Transaction is not expected to have any partitions since its state is "
                    + txnMetadata.txnState() + ": " + txnMetadata);
        }

        List<TransactionLogValue.PartitionsSchema> transactionPartitions = null;

        if (txnMetadata.txnState() != TransactionState.EMPTY) {
            transactionPartitions = txnMetadata.topicPartitions().stream()
                    .collect(Collectors.groupingBy(TopicPartition::topic))
                    .entrySet().stream()
                    .map(entry -> {
                        TransactionLogValue.PartitionsSchema schema = new TransactionLogValue.PartitionsSchema();
                        schema.setTopic(entry.getKey());
                        schema.setPartitionIds(
                                entry.getValue().stream()
                                        .map(TopicPartition::partition)
                                        .toList()
                        );
                        return schema;
                    }).toList();
        }

        return MessageUtil.toVersionPrefixedBytes(
                transactionVersionLevel.transactionLogValueVersion(),
                new TransactionLogValue()
                        .setProducerId(txnMetadata.producerId())
                        .setProducerEpoch(txnMetadata.producerEpoch())
                        .setTransactionTimeoutMs(txnMetadata.txnTimeoutMs())
                        .setTransactionStatus(txnMetadata.txnState().id())
                        .setTransactionLastUpdateTimestampMs(txnMetadata.txnLastUpdateTimestamp())
                        .setTransactionStartTimestampMs(txnMetadata.txnStartTimestamp())
                        .setTransactionPartitions(transactionPartitions)
                        .setClientTransactionVersion(txnMetadata.clientTransactionVersion().featureLevel())
        );
    }

    /**

Review comment (suggestion):
```java
                        new TransactionLogValue.PartitionsSchema()
                            .setTopic(entry.getKey())
                            .setPartitionIds(entry.getValue().stream().map(TopicPartition::partition).toList()
                        )
```

Code after:
    public static byte[] keyToBytes(String transactionalId) {
        return MessageUtil.toCoordinatorTypePrefixedBytes(
                new TransactionLogKey().setTransactionalId(transactionalId)
        );
    }

    /**
     * Generates the payload bytes for transaction log message value
     *
     * @return value payload bytes
     */
    public static byte[] valueToBytes(TxnTransitMetadata txnMetadata,
                                      TransactionVersion transactionVersionLevel) {
        if (txnMetadata.txnState() == TransactionState.EMPTY && !txnMetadata.topicPartitions().isEmpty()) {
            throw new IllegalStateException("Transaction is not expected to have any partitions since its state is "
                    + txnMetadata.txnState() + ": " + txnMetadata);
        }

        List<TransactionLogValue.PartitionsSchema> transactionPartitions = null;

        if (txnMetadata.txnState() != TransactionState.EMPTY) {
            transactionPartitions = txnMetadata.topicPartitions().stream()
                    .collect(Collectors.groupingBy(TopicPartition::topic))
                    .entrySet().stream()
                    .map(entry ->
                        new TransactionLogValue.PartitionsSchema().setTopic(entry.getKey())
                            .setPartitionIds(entry.getValue().stream().map(TopicPartition::partition).toList())).toList();
        }

        return MessageUtil.toVersionPrefixedBytes(
                transactionVersionLevel.transactionLogValueVersion(),
                new TransactionLogValue()
                        .setProducerId(txnMetadata.producerId())
                        .setProducerEpoch(txnMetadata.producerEpoch())
                        .setTransactionTimeoutMs(txnMetadata.txnTimeoutMs())
                        .setTransactionStatus(txnMetadata.txnState().id())
                        .setTransactionLastUpdateTimestampMs(txnMetadata.txnLastUpdateTimestamp())
                        .setTransactionStartTimestampMs(txnMetadata.txnStartTimestamp())
                        .setTransactionPartitions(transactionPartitions)
                        .setClientTransactionVersion(txnMetadata.clientTransactionVersion().featureLevel())
        );
    }

    /**
     * Decodes the transaction log messages' key
     *
     * @return the transactional id
     * @throws IllegalStateException if the version is not a valid transaction log key version
     */
    public static String readTxnRecordKey(ByteBuffer buffer) {
        short version = buffer.getShort();