MINOR; Revert "KAFKA-18681: Created GetReplicaLogInfo RPCs (#19664)" (#20371)

This reverts commit d86ba7f.

Reverting since we are planning to change how KIP-966 is implemented. We
should revert this RPC until we have more clarity on how this KIP will
be executed.

Reviewers: José Armando García Sancio <[email protected]>

Author: josefk31

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -134,8 +134,8 @@ public enum ApiKeys {
     STREAMS_GROUP_DESCRIBE(ApiMessageType.STREAMS_GROUP_DESCRIBE),
     DESCRIBE_SHARE_GROUP_OFFSETS(ApiMessageType.DESCRIBE_SHARE_GROUP_OFFSETS),
     ALTER_SHARE_GROUP_OFFSETS(ApiMessageType.ALTER_SHARE_GROUP_OFFSETS),
-    DELETE_SHARE_GROUP_OFFSETS(ApiMessageType.DELETE_SHARE_GROUP_OFFSETS),
-    GET_REPLICA_LOG_INFO(ApiMessageType.GET_REPLICA_LOG_INFO);
+    DELETE_SHARE_GROUP_OFFSETS(ApiMessageType.DELETE_SHARE_GROUP_OFFSETS);
+    
 
     private static final Map<ApiMessageType.ListenerType, EnumSet<ApiKeys>> APIS_BY_LISTENER =
         new EnumMap<>(ApiMessageType.ListenerType.class);

clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java

@@ -354,8 +354,6 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
                 return AlterShareGroupOffsetsRequest.parse(readable, apiVersion);
             case DELETE_SHARE_GROUP_OFFSETS:
                 return DeleteShareGroupOffsetsRequest.parse(readable, apiVersion);
-            case GET_REPLICA_LOG_INFO:
-                return GetReplicaLogInfoRequest.parse(readable, apiVersion);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java

@@ -291,8 +291,6 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Readable readable,
                 return AlterShareGroupOffsetsResponse.parse(readable, version);
             case DELETE_SHARE_GROUP_OFFSETS:
                 return DeleteShareGroupOffsetsResponse.parse(readable, version);
-            case GET_REPLICA_LOG_INFO:
-                return GetReplicaLogInfoResponse.parse(readable, version);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/GetReplicaLogInfoRequest.java

@@ -157,8 +157,6 @@
 import org.apache.kafka.common.message.FetchSnapshotRequestData;
 import org.apache.kafka.common.message.FetchSnapshotResponseData;
 import org.apache.kafka.common.message.FindCoordinatorRequestData;
-import org.apache.kafka.common.message.GetReplicaLogInfoRequestData;
-import org.apache.kafka.common.message.GetReplicaLogInfoResponseData;
 import org.apache.kafka.common.message.GetTelemetrySubscriptionsRequestData;
 import org.apache.kafka.common.message.GetTelemetrySubscriptionsResponseData;
 import org.apache.kafka.common.message.HeartbeatRequestData;
@@ -1077,7 +1075,6 @@ private AbstractRequest getRequest(ApiKeys apikey, short version) {
             case DESCRIBE_SHARE_GROUP_OFFSETS: return createDescribeShareGroupOffsetsRequest(version);
             case ALTER_SHARE_GROUP_OFFSETS: return createAlterShareGroupOffsetsRequest(version);
             case DELETE_SHARE_GROUP_OFFSETS: return createDeleteShareGroupOffsetsRequest(version);
-            case GET_REPLICA_LOG_INFO: return createGetReplicaLogInfoRequest(version);
             default: throw new IllegalArgumentException("Unknown API key " + apikey);
         }
     }
@@ -1173,25 +1170,10 @@ private AbstractResponse getResponse(ApiKeys apikey, short version) {
             case DESCRIBE_SHARE_GROUP_OFFSETS: return createDescribeShareGroupOffsetsResponse();
             case ALTER_SHARE_GROUP_OFFSETS: return createAlterShareGroupOffsetsResponse();
             case DELETE_SHARE_GROUP_OFFSETS: return createDeleteShareGroupOffsetsResponse();
-            case GET_REPLICA_LOG_INFO: return createGetReplicaLogInfoResponse();
             default: throw new IllegalArgumentException("Unknown API key " + apikey);
         }
     }
 
-    private GetReplicaLogInfoRequest createGetReplicaLogInfoRequest(short version) {
-        GetReplicaLogInfoRequestData data = new GetReplicaLogInfoRequestData()
-                .setTopicPartitions(singletonList(new GetReplicaLogInfoRequestData.TopicPartitions()
-                .setPartitions(singletonList(0))));
-        return new GetReplicaLogInfoRequest.Builder(data).build(version);
-    }
-
-    private GetReplicaLogInfoResponse createGetReplicaLogInfoResponse() {
-        GetReplicaLogInfoResponseData data = new GetReplicaLogInfoResponseData();
-        data.setBrokerEpoch(0);
-        data.setTopicPartitionLogInfoList(singletonList(new GetReplicaLogInfoResponseData.TopicPartitionLogInfo()));
-        return new GetReplicaLogInfoResponse(data);
-    }
-
     private ConsumerGroupDescribeRequest createConsumerGroupDescribeRequest(short version) {
         ConsumerGroupDescribeRequestData data = new ConsumerGroupDescribeRequestData()
             .setGroupIds(Collections.singletonList("group"))

clients/src/main/java/org/apache/kafka/common/requests/GetReplicaLogInfoResponse.java

@@ -44,7 +44,6 @@
 
 import java.util.Map;
 import java.util.Optional;
-import java.util.function.Supplier;
 
 import scala.jdk.javaapi.OptionConverters;
 
@@ -72,7 +71,6 @@ public class KafkaApisBuilder {
     private ClientMetricsManager clientMetricsManager = null;
     private ShareCoordinator shareCoordinator = null;
     private GroupConfigManager groupConfigManager = null;
-    private Supplier<Long> brokerEpochSupplier = () -> -1L;
 
     public KafkaApisBuilder setRequestChannel(RequestChannel requestChannel) {
         this.requestChannel = requestChannel;
@@ -189,11 +187,6 @@ public KafkaApisBuilder setGroupConfigManager(GroupConfigManager groupConfigMana
         return this;
     }
 
-    public KafkaApisBuilder setBrokerEpochSupplier(Supplier<Long> brokerEpochSupplier) {
-        this.brokerEpochSupplier = brokerEpochSupplier;
-        return this;
-    }
-
     @SuppressWarnings({"CyclomaticComplexity"})
     public KafkaApis build() {
         if (requestChannel == null) throw new RuntimeException("you must set requestChannel");
@@ -237,7 +230,6 @@ public KafkaApis build() {
                              tokenManager,
                              apiVersionManager,
                              clientMetricsManager,
-                             groupConfigManager,
-                             brokerEpochSupplier);
+                             groupConfigManager);
     }
 }

clients/src/main/resources/common/message/GetReplicaLogInfoRequest.json

@@ -463,8 +463,7 @@ class BrokerServer(
         tokenManager = tokenManager,
         apiVersionManager = apiVersionManager,
         clientMetricsManager = clientMetricsManager,
-        groupConfigManager = groupConfigManager,
-        brokerEpochSupplier = () => lifecycleManager.brokerEpoch)
+        groupConfigManager = groupConfigManager)
 
       dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.nodeId,
         socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,