KAFKA-16557 Implemented OffsetFetchRequestState toStringBase and added a test for it in CommitRequestManagerTest

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

@@ -880,6 +880,11 @@ boolean retryTimeoutExpired(long currentTimeMs) {
         }
 
         abstract void removeRequest();
+
+        // Visible for testing
+        protected Optional<Long> expirationTimeMs() {
+            return expirationTimeMs;
+        }
     }
 
     class OffsetFetchRequestState extends RetriableRequestState {
@@ -1090,14 +1095,13 @@ private void chainFuture(
         }
 
         @Override
-        public String toString() {
-            return "OffsetFetchRequestState{" +
-                    "requestedPartitions=" + requestedPartitions +
-                    ", memberId=" + memberInfo.memberId.orElse("undefined") +
-                    ", memberEpoch=" + (memberInfo.memberEpoch.isPresent() ? memberInfo.memberEpoch.get() : "undefined") +
-                    ", future=" + future +
-                    ", " + toStringBase() +
-                    '}';
+        public String toStringBase() {
+            return super.toStringBase() +
+                    ", memberInfo=" + memberInfo +
+                    ", expirationTimeMs=" + (expirationTimeMs().isPresent() ? expirationTimeMs() : "undefined") +
+                    ", isExpired=" + isExpired +
+                    ", requestedPartitions=" + requestedPartitions +
+                    ", future=" + future;
         }
     }
 
@@ -1289,5 +1293,11 @@ static class MemberInfo {
             this.memberId = Optional.empty();
             this.memberEpoch = Optional.empty();
         }
+
+        @Override
+        public String toString() {
+            return "MemberInfo{" + "memberId=" + memberId.orElse("undefined") +
+                    ", memberEpoch=" + (memberEpoch.isPresent() ? memberEpoch : "undefined") + "}";
+        }
     }
 }

clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java

@@ -76,6 +76,7 @@
 import static org.apache.kafka.clients.consumer.internals.ConsumerTestBuilder.DEFAULT_GROUP_ID;
 import static org.apache.kafka.clients.consumer.internals.ConsumerTestBuilder.DEFAULT_GROUP_INSTANCE_ID;
 import static org.apache.kafka.test.TestUtils.assertFutureThrows;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -123,6 +124,50 @@ public void setup() {
         this.props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
     }
 
+    @Test
+    public void testOffsetFetchRequestStateToStringBase() {
+        ConsumerConfig config = mock(ConsumerConfig.class);
+        CommitRequestManager.MemberInfo memberInfo = new CommitRequestManager.MemberInfo();
+
+        CommitRequestManager commitRequestManager = new CommitRequestManager(
+                time,
+                logContext,
+                subscriptionState,
+                config,
+                coordinatorRequestManager,
+                offsetCommitCallbackInvoker,
+                "groupId",
+                Optional.of("groupInstanceId"),
+                metrics);
+
+        Set<TopicPartition> requestedPartitions = new HashSet<>();
+        TopicPartition topicPartition1 = new TopicPartition("topic-1", 1);
+        requestedPartitions.add(topicPartition1);
+
+        CommitRequestManager.OffsetFetchRequestState offsetFetchRequestState = commitRequestManager.new OffsetFetchRequestState(
+                requestedPartitions,
+                retryBackoffMs,
+                retryBackoffMaxMs,
+                1000,
+                memberInfo);
+
+        RequestState requestState = new RequestState(
+                logContext,
+                "CommitRequestManager",
+                retryBackoffMs,
+                retryBackoffMaxMs);
+
+        String target = requestState.toStringBase() +
+                ", memberInfo=" + memberInfo +
+                ", expirationTimeMs=" + (offsetFetchRequestState.expirationTimeMs().isPresent() ? offsetFetchRequestState.expirationTimeMs() : "undefined") +
+                ", isExpired=" + offsetFetchRequestState.isExpired +
+                ", requestedPartitions=" + offsetFetchRequestState.requestedPartitions +
+                ", future=" + offsetFetchRequestState.future();
+
+        assertDoesNotThrow(requestState::toString);
+        assertEquals(target, offsetFetchRequestState.toStringBase());
+    }
+
     @Test
     public void testPollSkipIfCoordinatorUnknown() {
         CommitRequestManager commitRequestManager = create(false, 0);