From 09ca266533db9f0d28974be31fefa5cae26cc32a Mon Sep 17 00:00:00 2001
From: Mitsuaki Ito <itoumagic@gmail.com>
Date: Mon, 6 Jun 2022 15:12:58 +0900
Subject: [PATCH] fix(topicdata): handle unsupported DescribeLogDirs  for MSK
 Serverless (#1113)

close #1112
---
 src/main/java/org/akhq/modules/AbstractKafkaWrapper.java | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/main/java/org/akhq/modules/AbstractKafkaWrapper.java b/src/main/java/org/akhq/modules/AbstractKafkaWrapper.java
index 2e9e9bf8b..3aaf3b62d 100644
--- a/src/main/java/org/akhq/modules/AbstractKafkaWrapper.java
+++ b/src/main/java/org/akhq/modules/AbstractKafkaWrapper.java
@@ -16,6 +16,7 @@
 import org.apache.kafka.common.errors.ClusterAuthorizationException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 
 import java.util.*;
 import java.util.concurrent.ExecutionException;
@@ -254,7 +255,7 @@ public Map<Integer, Map<String, LogDirDescription>> describeLogDir(String cluste
                             .allDescriptions()
                             .get();
                     } catch (ExecutionException e) {
-                        if (e.getCause() instanceof ClusterAuthorizationException || e.getCause() instanceof TopicAuthorizationException) {
+                        if (e.getCause() instanceof ClusterAuthorizationException || e.getCause() instanceof TopicAuthorizationException || e.getCause() instanceof UnsupportedVersionException) {
                             return new HashMap<>();
                         }