hachikuji commented on a change in pull request #10206:
URL: https://github.com/apache/kafka/pull/10206#discussion_r582565318
##########
File path:
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
##########
@@ -223,6 +224,46 @@ class TransactionStateManager(brokerId: Int,
throw new IllegalStateException(s"Unexpected empty transaction metadata
returned while putting $txnMetadata")))
}
+ def listTransactionStates(
+ filterProducerIds: Set[Long],
+ filterStateNames: Set[String]
+ ): Either[Errors, List[ListTransactionsResponseData.TransactionState]] = {
+ inReadLock(stateLock) {
+ if (loadingPartitions.nonEmpty) {
+ Left(Errors.COORDINATOR_LOAD_IN_PROGRESS)
+ } else {
+ val filterStates = filterStateNames.flatMap(TransactionState.fromName)
Review comment:
I could be persuaded probably. Suppose we add a new state in the future.
If a user tried to query that state on an old broker, definitely no
transactions would exist in that state. From that perspective, the
implementation would return an accurate result. On the other hand, I can see
how that might be misleading. If we decide to return an error code, then I
think we'll have to treat the state filter a little more carefully from a
versioning perspective. A new state would probably demand a new version. I
think we tried to view the states a little bit more loosely in the `ListGroups`
API (which this is modeled after) because we considered the state machine more
of an internal implementation detail. This, by the way, is why we went with the
string representation rather than numeric ids. This became more of a grey area
though when the state filter was added to the request...
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
