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

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

@@ -1078,14 +1078,11 @@ 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 +
+                    ", requestedPartitions=" + requestedPartitions +
+                    ", future=" + future;
         }
     }
 
@@ -1278,5 +1275,12 @@ 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

@@ -75,6 +75,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;
@@ -122,6 +123,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);
+
+        TimedRequestState timedRequestState = new TimedRequestState(
+                logContext,
+                "CommitRequestManager",
+                retryBackoffMs,
+                retryBackoffMaxMs,
+                TimedRequestState.deadlineTimer(time, 0)
+        );
+
+        String target = timedRequestState.toStringBase() +
+                ", memberInfo=" + memberInfo +
+                ", requestedPartitions=" + offsetFetchRequestState.requestedPartitions +
+                ", future=" + offsetFetchRequestState.future();
+
+        assertDoesNotThrow(timedRequestState::toString);
+        assertEquals(target, offsetFetchRequestState.toStringBase());
+    }
+
     @Test
     public void testPollSkipIfCoordinatorUnknown() {
         CommitRequestManager commitRequestManager = create(false, 0);