Minor updates to naming

Author: kirktrue

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

@@ -171,7 +171,7 @@ public class TransactionManager {
      *
      * See KAFKA-14831 for more detail.
      */
-    public interface InvalidTransitionAttemptPolicy {
+    public interface InvalidStateTransitionAttemptHandler {
 
         /**
          * Callback to determine if the transaction manager's {@link #currentState} should be set to
@@ -184,11 +184,11 @@ public interface InvalidTransitionAttemptPolicy {
     }
 
     /**
-     * The default policy is to poison the internal {@link #currentState state} if an invalid state transition is
+     * The default logic is to poison the internal {@link #currentState state} if an invalid state transition is
      * attempted on the {@link Sender.SenderThread sender's thread}.
      */
-    public static final InvalidTransitionAttemptPolicy DEFAULT_INVALID_TRANSITION_ATTEMPT_POLICY = () -> Thread.currentThread() instanceof Sender.SenderThread;
-    private final InvalidTransitionAttemptPolicy invalidTransitionAttemptPolicy;
+    public static final InvalidStateTransitionAttemptHandler DEFAULT_INVALID_STATE_TRANSITION_ATTEMPT_HANDLER = () -> Thread.currentThread() instanceof Sender.SenderThread;
+    private final InvalidStateTransitionAttemptHandler invalidStateTransitionAttemptHandler;
 
     private PendingStateTransition pendingTransition;
 
@@ -274,15 +274,15 @@ public TransactionManager(final LogContext logContext,
                               final int transactionTimeoutMs,
                               final long retryBackoffMs,
                               final ApiVersions apiVersions) {
-        this(logContext, transactionalId, transactionTimeoutMs, retryBackoffMs, apiVersions, DEFAULT_INVALID_TRANSITION_ATTEMPT_POLICY);
+        this(logContext, transactionalId, transactionTimeoutMs, retryBackoffMs, apiVersions, DEFAULT_INVALID_STATE_TRANSITION_ATTEMPT_HANDLER);
     }
 
     public TransactionManager(final LogContext logContext,
                               final String transactionalId,
                               final int transactionTimeoutMs,
                               final long retryBackoffMs,
                               final ApiVersions apiVersions,
-                              final InvalidTransitionAttemptPolicy invalidTransitionAttemptPolicy) {
+                              final InvalidStateTransitionAttemptHandler invalidStateTransitionAttemptHandler) {
         this.producerIdAndEpoch = ProducerIdAndEpoch.NONE;
         this.transactionalId = transactionalId;
         this.log = logContext.logger(TransactionManager.class);
@@ -299,7 +299,7 @@ public TransactionManager(final LogContext logContext,
         this.retryBackoffMs = retryBackoffMs;
         this.txnPartitionMap = new TxnPartitionMap(logContext);
         this.apiVersions = apiVersions;
-        this.invalidTransitionAttemptPolicy = invalidTransitionAttemptPolicy;
+        this.invalidStateTransitionAttemptHandler = invalidStateTransitionAttemptHandler;
     }
 
     public synchronized TransactionalRequestResult initializeTransactions() {
@@ -1086,7 +1086,7 @@ private void transitionTo(State target, RuntimeException error) {
             String message = idString + "Invalid transition attempted from state "
                     + currentState.name() + " to state " + target.name();
 
-            if (invalidTransitionAttemptPolicy.isFatalState()) {
+            if (invalidStateTransitionAttemptHandler.isFatalState()) {
                 currentState = State.FATAL_ERROR;
                 lastError = new IllegalStateException(message);
                 throw lastError;

clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java

@@ -163,12 +163,12 @@ public void setup() {
     }
 
     private void initializeTransactionManager(Optional<String> transactionalId, boolean transactionV2Enabled) {
-        initializeTransactionManager(transactionalId, transactionV2Enabled, TransactionManager.DEFAULT_INVALID_TRANSITION_ATTEMPT_POLICY);
+        initializeTransactionManager(transactionalId, transactionV2Enabled, TransactionManager.DEFAULT_INVALID_STATE_TRANSITION_ATTEMPT_HANDLER);
     }
 
     private void initializeTransactionManager(Optional<String> transactionalId,
                                               boolean transactionV2Enabled,
-                                              TransactionManager.InvalidTransitionAttemptPolicy invalidTransitionAttemptPolicy) {
+                                              TransactionManager.InvalidStateTransitionAttemptHandler invalidStateTransitionAttemptHandler) {
         Metrics metrics = new Metrics(time);
 
         apiVersions.update("0", new NodeApiVersions(Arrays.asList(
@@ -195,7 +195,7 @@ private void initializeTransactionManager(Optional<String> transactionalId,
             finalizedFeaturesEpoch));
         finalizedFeaturesEpoch += 1;
         this.transactionManager = new TransactionManager(logContext, transactionalId.orElse(null),
-                transactionTimeoutMs, DEFAULT_RETRY_BACKOFF_MS, apiVersions, invalidTransitionAttemptPolicy);
+                transactionTimeoutMs, DEFAULT_RETRY_BACKOFF_MS, apiVersions, invalidStateTransitionAttemptHandler);
 
         int batchSize = 16 * 1024;
         int deliveryTimeoutMs = 3000;
@@ -3805,10 +3805,11 @@ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(new Metrics(t
 
     @Test
     public void testBackgroundInvalidStateTransitionIsFatal() {
-        // Usually the policy is the poison the transaction manager only by the Sender thread, but for the
-        // test the policy is forced to true to mimic it being called from the Sender.
-        TransactionManager.InvalidTransitionAttemptPolicy policy = () -> true;
-        initializeTransactionManager(Optional.of(transactionalId), true, policy);
+        // The default logic is to poison the transaction manager on an invalid state transition attempt if that
+        // attempt happens on the Sender thread. Here the logic is altered to always poison on invalid attempts to
+        // mimic as though it's being invoked by the Sender.
+        TransactionManager.InvalidStateTransitionAttemptHandler testHandler = () -> true;
+        initializeTransactionManager(Optional.of(transactionalId), true, testHandler);
         doInitTransactions();
         assertTrue(transactionManager.isTransactional());