KAFKA-18782: Extend ApplicationRecoverableException related exceptions (#19354)

**Summary**
Extend ApplicationRecoverableException related exceptions

Reviewers: Artem Livshits <alivshits@confluent.io>, Justine Olshan
 <jolshan@confluent.io>

clients/src/main/java/org/apache/kafka/common/errors/ApplicationRecoverableException.java

@@ -22,7 +22,22 @@ package org.apache.kafka.common.errors;
  * different recovery strategies (e.g., re-balancing task, restoring from checkpoints) may be employed.
  */
 public abstract class ApplicationRecoverableException extends ApiException {
+    private static final long serialVersionUID = 1L;
+
+    public ApplicationRecoverableException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
     public ApplicationRecoverableException(String message) {
         super(message);
     }
+
+    public ApplicationRecoverableException(Throwable cause) {
+        super(cause);
+    }
+
+    public ApplicationRecoverableException() {
+        super();
+    }
+
 }

clients/src/main/java/org/apache/kafka/common/errors/FencedInstanceIdException.java

@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.errors;
 
-public class FencedInstanceIdException extends ApiException {
+public class FencedInstanceIdException extends ApplicationRecoverableException {
     private static final long serialVersionUID = 1L;
 
     public FencedInstanceIdException(String message) {

clients/src/main/java/org/apache/kafka/common/errors/IllegalGenerationException.java

@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.errors;
 
-public class IllegalGenerationException extends ApiException {
+public class IllegalGenerationException extends ApplicationRecoverableException {
     private static final long serialVersionUID = 1L;
 
     public IllegalGenerationException() {

clients/src/main/java/org/apache/kafka/common/errors/InvalidPidMappingException.java

@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.errors;
 
-public class InvalidPidMappingException extends ApiException {
+public class InvalidPidMappingException extends ApplicationRecoverableException {
     public InvalidPidMappingException(String message) {
         super(message);
     }

clients/src/main/java/org/apache/kafka/common/errors/InvalidProducerEpochException.java

@@ -22,7 +22,7 @@ package org.apache.kafka.common.errors;
  * by calling KafkaProducer#abortTransaction which would try to send initPidRequest and reinitialize the producer
  * under the hood.
  */
-public class InvalidProducerEpochException extends ApiException {
+public class InvalidProducerEpochException extends ApplicationRecoverableException {
 
     private static final long serialVersionUID = 1L;
 

clients/src/main/java/org/apache/kafka/common/errors/ProducerFencedException.java

@@ -22,7 +22,7 @@ package org.apache.kafka.common.errors;
  * given time, and the latest one to be started "fences" the previous instances so that they can no longer
  * make transactional requests. When you encounter this exception, you must close the producer instance.
  */
-public class ProducerFencedException extends ApiException {
+public class ProducerFencedException extends ApplicationRecoverableException {
 
     public ProducerFencedException(String msg) {
         super(msg);

clients/src/main/java/org/apache/kafka/common/errors/UnknownMemberIdException.java

@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.errors;
 
-public class UnknownMemberIdException extends ApiException {
+public class UnknownMemberIdException extends ApplicationRecoverableException {
     private static final long serialVersionUID = 1L;
 
     public UnknownMemberIdException() {

clients/src/test/java/org/apache/kafka/common/errors/TransactionExceptionHierarchyTest.java

@@ -75,4 +75,23 @@ public class TransactionExceptionHierarchyTest {
         assertTrue(RefreshRetriableException.class.isAssignableFrom(exceptionClass),
                 exceptionClass.getSimpleName() + " should extend RefreshRetriableException");
     }
+
+    /**
+     * Verifies that the given exception class extends `ApplicationRecoverableException`
+     *
+     * @param exceptionClass the exception class to check
+     */
+    @ParameterizedTest
+    @ValueSource(classes = {
+        FencedInstanceIdException.class,
+        IllegalGenerationException.class,
+        InvalidPidMappingException.class,
+        InvalidProducerEpochException.class,
+        ProducerFencedException.class,
+        UnknownMemberIdException.class
+    })
+    void testApplicationRecoverableExceptionHierarchy(Class<? extends Exception> exceptionClass) {
+        assertTrue(ApplicationRecoverableException.class.isAssignableFrom(exceptionClass),
+                exceptionClass.getSimpleName() + " should extend ApplicationRecoverableException");
+    }
 }