Merge branch 'main' into fix-null-value-internal-topic

germanosin · web-flow · commit 92e3d43b0040 · 2025-11-06T09:02:55.000+01:00
diff --git a/api/src/main/java/io/kafbat/ui/service/AdminClientService.java b/api/src/main/java/io/kafbat/ui/service/AdminClientService.java
@@ -3,8 +3,10 @@
 import io.kafbat.ui.model.KafkaCluster;
 import reactor.core.publisher.Mono;
 
-public interface AdminClientService {
+public interface AdminClientService extends AutoCloseable {
 
   Mono<ReactiveAdminClient> get(KafkaCluster cluster);
 
+  void invalidate(KafkaCluster cluster, Throwable e);
+
 }
diff --git a/api/src/main/java/io/kafbat/ui/service/AdminClientServiceImpl.java b/api/src/main/java/io/kafbat/ui/service/AdminClientServiceImpl.java
@@ -19,7 +19,7 @@
 
 @Service
 @Slf4j
-public class AdminClientServiceImpl implements AdminClientService, Closeable {
+public class AdminClientServiceImpl implements AdminClientService {
 
   private static final int DEFAULT_CLIENT_TIMEOUT_MS = 30_000;
 
@@ -62,6 +62,21 @@ private Mono<ReactiveAdminClient> createAdminClient(KafkaCluster cluster) {
             "Error while creating AdminClient for the cluster " + cluster.getName(), th));
   }
 
+  @Override
+  public void invalidate(KafkaCluster cluster, Throwable e) {
+    if (e.getClass().getCanonicalName().startsWith("org.apache.kafka.common.errors")) {
+      log.warn("AdminClient for the cluster {} is invalidated due to {}", cluster.getName(), e.getMessage());
+      ReactiveAdminClient client = adminClientCache.remove(cluster.getName());
+      if (client != null) {
+        try {
+          client.close();
+        } catch (Exception ce) {
+          log.info("Error while closing AdminClient for the cluster {}", cluster.getName(), ce);
+        }
+      }
+    }
+  }
+
   @Override
   public void close() {
     adminClientCache.values().forEach(ReactiveAdminClient::close);
diff --git a/api/src/main/java/io/kafbat/ui/service/BrokerService.java b/api/src/main/java/io/kafbat/ui/service/BrokerService.java
@@ -73,7 +73,8 @@ public Flux<InternalBroker> getBrokers(KafkaCluster cluster) {
         .map(description -> description.getNodes().stream()
             .map(node -> new InternalBroker(node, partitionsDistribution, stats.getMetrics()))
             .collect(Collectors.toList()))
-        .flatMapMany(Flux::fromIterable);
+        .flatMapMany(Flux::fromIterable)
+        .doOnError(e -> adminClientService.invalidate(cluster, e));
   }
 
   public Mono<Void> updateBrokerLogDir(KafkaCluster cluster,
diff --git a/api/src/main/java/io/kafbat/ui/service/StatisticsService.java b/api/src/main/java/io/kafbat/ui/service/StatisticsService.java
@@ -49,7 +49,8 @@ private Mono<Statistics> getStatistics(KafkaCluster cluster) {
                         )
                     )
             ).doOnError(e ->
-                log.error("Failed to collect cluster {} info", cluster.getName(), e))
+                log.error("Failed to collect cluster {} info", cluster.getName(), e)
+            ).doOnError(e -> adminClientService.invalidate(cluster, e))
             .onErrorResume(t -> Mono.just(Statistics.statsUpdateError(t))));
   }
 
diff --git a/api/src/main/java/io/kafbat/ui/service/TopicsService.java b/api/src/main/java/io/kafbat/ui/service/TopicsService.java
@@ -169,7 +169,8 @@ public Mono<List<ConfigEntry>> getTopicConfigs(KafkaCluster cluster, String topi
         .flatMap(ac -> ac.describeTopic(topicName)
             .switchIfEmpty(Mono.error(new TopicNotFoundException()))
             .then(ac.getTopicsConfig(List.of(topicName), true))
-            .map(m -> m.values().stream().findFirst().orElse(List.of())));
+            .map(m -> m.values().stream().findFirst().orElse(List.of())))
+            .doOnError(e -> adminClientService.invalidate(cluster, e));
   }
 
   private Mono<InternalTopic> createTopic(KafkaCluster c, ReactiveAdminClient adminClient, TopicCreationDTO topicData) {
@@ -178,6 +179,7 @@ private Mono<InternalTopic> createTopic(KafkaCluster c, ReactiveAdminClient admi
             topicData.getPartitions(),
             topicData.getReplicationFactor(),
             topicData.getConfigs())
+        .doOnError(e -> adminClientService.invalidate(c, e))
         .thenReturn(topicData)
         .onErrorMap(t -> new TopicMetadataException(t.getMessage(), t))
         .then(loadTopicAfterCreation(c, topicData.getName()));
@@ -492,7 +494,8 @@ private Mono<List<InternalTopic>> filterExisting(KafkaCluster cluster, Collectio
         .map(existing -> topics
             .stream()
             .filter(s -> existing.contains(s.getName()))
-            .collect(toList()));
+            .collect(toList())
+        ).doOnError(e -> adminClientService.invalidate(cluster, e));
   }
 
 }
diff --git a/api/src/test/java/io/kafbat/ui/mapper/KafkaConnectMapperTest.java b/api/src/test/java/io/kafbat/ui/mapper/KafkaConnectMapperTest.java
@@ -42,8 +42,13 @@ void toKafkaConnect() {
 
       ConnectorDTO connectorDto = new ConnectorDTO();
       connectorDto.setName(UUID.randomUUID().toString());
+
+      String traceMessage = connectorState == ConnectorStateDTO.FAILED
+          ? "Test error trace for failed connector"
+          : null;
+
       connectorDto.setStatus(
-          new ConnectorStatusDTO(connectorState, UUID.randomUUID().toString())
+          new ConnectorStatusDTO(connectorState, UUID.randomUUID().toString(), traceMessage)
       );
 
       List<TaskDTO> tasks = new ArrayList<>();
diff --git a/api/src/test/java/io/kafbat/ui/service/AdminClientServiceTest.java b/api/src/test/java/io/kafbat/ui/service/AdminClientServiceTest.java
@@ -0,0 +1,25 @@
+package io.kafbat.ui.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.kafbat.ui.AbstractIntegrationTest;
+import io.kafbat.ui.model.KafkaCluster;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+class AdminClientServiceTest extends AbstractIntegrationTest {
+  @Test
+  void testInvalidateOnFailure() {
+    AdminClientService adminClientService = applicationContext.getBean(AdminClientService.class);
+    ClustersStorage clustersStorage = applicationContext.getBean(ClustersStorage.class);
+    KafkaCluster cluster = clustersStorage.getClusterByName(LOCAL).get();
+    ReactiveAdminClient clientBefore = adminClientService.get(cluster).block();
+    ReactiveAdminClient clientBeforeRepeat = adminClientService.get(cluster).block();
+    assertThat(clientBeforeRepeat).isEqualTo(clientBefore);
+    adminClientService.invalidate(cluster, new TimeoutException());
+    ReactiveAdminClient clientAfter = adminClientService.get(cluster).block();
+    assertThat(clientAfter).isNotEqualTo(clientBefore);
+  }
+}
diff --git a/api/src/test/java/io/kafbat/ui/service/index/KafkaConnectNgramFilterTest.java b/api/src/test/java/io/kafbat/ui/service/index/KafkaConnectNgramFilterTest.java
@@ -14,25 +14,22 @@ class KafkaConnectNgramFilterTest extends AbstractNgramFilterTest<FullConnectorI
 
   @Override
   protected NgramFilter<FullConnectorInfoDTO> buildFilter(List<FullConnectorInfoDTO> items,
-                                                          boolean enabled,
-                                                          ClustersProperties.NgramProperties ngramProperties) {
+      boolean enabled,
+      ClustersProperties.NgramProperties ngramProperties) {
     return new KafkaConnectNgramFilter(items, enabled, ngramProperties);
   }
 
   @Override
   protected List<FullConnectorInfoDTO> items() {
-    return IntStream.range(0, 100).mapToObj(i ->
-        new FullConnectorInfoDTO(
-            "connect-" + i,
-            "connector-" + i,
-            "class",
-            ConnectorTypeDTO.SINK,
-            List.of(),
-            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, "reason"),
-            1,
-            0
-        )
-    ).toList();
+    return IntStream.range(0, 100).mapToObj(i -> new FullConnectorInfoDTO(
+        "connect-" + i,
+        "connector-" + i,
+        "class",
+        ConnectorTypeDTO.SINK,
+        List.of(),
+        new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, "worker-1", "reason"),
+        1,
+        0)).toList();
   }
 
   @Override
@@ -55,7 +52,7 @@ protected List<FullConnectorInfoDTO> sortedItems() {
             "class",
             ConnectorTypeDTO.SINK,
             List.of(),
-            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, "reason"),
+            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, null, "reason"),
             1,
             0
         ),
@@ -65,7 +62,7 @@ protected List<FullConnectorInfoDTO> sortedItems() {
             "class",
             ConnectorTypeDTO.SINK,
             List.of(),
-            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, "reason"),
+            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, null, "reason"),
             1,
             0
         )
@@ -86,20 +83,17 @@ protected List<FullConnectorInfoDTO> sortedResult(List<FullConnectorInfoDTO> ite
             "class",
             ConnectorTypeDTO.SINK,
             List.of(),
-            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, "reason"),
+            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, null, "reason"),
             1,
-            0
-        ),
+            0),
         new FullConnectorInfoDTO(
             "connect-pay",
             "connector-pay",
             "class",
             ConnectorTypeDTO.SINK,
             List.of(),
-            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, "reason"),
+            new ConnectorStatusDTO(ConnectorStateDTO.RUNNING, null, "reason"),
             1,
-            0
-        )
-    );
+            0));
   }
 }
diff --git a/contract-typespec/api/kafka-connect.tsp b/contract-typespec/api/kafka-connect.tsp
@@ -217,6 +217,7 @@ enum ConnectorState {
 model ConnectorStatus {
   state: ConnectorState;
   workerId?: string;
+  trace?: string;
 }
 
 model Connector {
diff --git a/contract/src/main/resources/swagger/kafbat-ui-api.yaml b/contract/src/main/resources/swagger/kafbat-ui-api.yaml
@@ -3756,6 +3756,8 @@ components:
           $ref: '#/components/schemas/ConnectorState'
         workerId:
           type: string
+        trace:
+          type: string
       required:
         - state
 
diff --git a/frontend/src/components/Connect/Details/Overview/Overview.styled.tsx b/frontend/src/components/Connect/Details/Overview/Overview.styled.tsx
@@ -0,0 +1,25 @@
+import styled, { css } from 'styled-components';
+
+export const WorkerInfo = styled.p(
+  ({ theme: { modal } }) => css`
+    margin: 4px 0 0 0;
+    font-size: 14px;
+    color: ${modal.contentColor};
+  `
+);
+
+export const TraceContent = styled.div(
+  ({ theme: { modal } }) => css`
+    background-color: ${modal.border.contrast};
+    padding: 16px;
+    border-radius: 6px;
+    font-family: 'Monaco', 'Menlo', 'Ubuntu Mono', monospace;
+    font-size: 12px;
+    color: ${modal.color};
+    border: 1px solid ${modal.border.contrast};
+    max-height: 400px;
+    overflow-y: auto;
+    white-space: pre-wrap;
+    word-break: break-word;
+  `
+);
diff --git a/frontend/src/components/Connect/Details/Overview/Overview.tsx b/frontend/src/components/Connect/Details/Overview/Overview.tsx
@@ -1,15 +1,20 @@
-import React from 'react';
+import React, { useState } from 'react';
 import * as C from 'components/common/Tag/Tag.styled';
 import * as Metrics from 'components/common/Metrics';
+import { Button } from 'components/common/Button/Button';
+import { Modal } from 'components/common/Modal';
 import getTagColor from 'components/common/Tag/getTagColor';
 import { RouterParamsClusterConnectConnector } from 'lib/paths';
 import useAppParams from 'lib/hooks/useAppParams';
 import { useConnector, useConnectorTasks } from 'lib/hooks/api/kafkaConnect';
+import { ConnectorState, Connector } from 'generated-sources';
 
 import getTaskMetrics from './getTaskMetrics';
+import * as S from './Overview.styled';
 
 const Overview: React.FC = () => {
   const routerProps = useAppParams<RouterParamsClusterConnectConnector>();
+  const [showTraceModal, setShowTraceModal] = useState(false);
 
   const { data: connector } = useConnector(routerProps);
   const { data: tasks } = useConnectorTasks(routerProps);
@@ -20,35 +25,78 @@ const Overview: React.FC = () => {
 
   const { running, failed } = getTaskMetrics(tasks);
 
+  const canShowTrace = (connectorData: Connector) =>
+    connectorData.status.state === ConnectorState.FAILED &&
+    !!connectorData.status.trace;
+
+  const handleStateClick = () => {
+    if (canShowTrace(connector)) {
+      setShowTraceModal(true);
+    }
+  };
+
   return (
-    <Metrics.Wrapper>
-      <Metrics.Section>
-        {connector.status?.workerId && (
-          <Metrics.Indicator label="Worker">
-            {connector.status.workerId}
+    <>
+      <Metrics.Wrapper>
+        <Metrics.Section>
+          {connector.status?.workerId && (
+            <Metrics.Indicator label="Worker">
+              {connector.status.workerId}
+            </Metrics.Indicator>
+          )}
+          <Metrics.Indicator label="Type">{connector.type}</Metrics.Indicator>
+          {connector.config['connector.class'] && (
+            <Metrics.Indicator label="Class">
+              {connector.config['connector.class']}
+            </Metrics.Indicator>
+          )}
+          <Metrics.Indicator label="State">
+            <C.Tag
+              color={getTagColor(connector.status.state)}
+              clickable={canShowTrace(connector)}
+              onClick={handleStateClick}
+            >
+              {connector.status.state}
+            </C.Tag>
           </Metrics.Indicator>
-        )}
-        <Metrics.Indicator label="Type">{connector.type}</Metrics.Indicator>
-        {connector.config['connector.class'] && (
-          <Metrics.Indicator label="Class">
-            {connector.config['connector.class']}
+          <Metrics.Indicator label="Tasks Running">{running}</Metrics.Indicator>
+          <Metrics.Indicator
+            label="Tasks Failed"
+            isAlert
+            alertType={failed > 0 ? 'error' : 'success'}
+          >
+            {failed}
           </Metrics.Indicator>
-        )}
-        <Metrics.Indicator label="State">
-          <C.Tag color={getTagColor(connector.status.state)}>
-            {connector.status.state}
-          </C.Tag>
-        </Metrics.Indicator>
-        <Metrics.Indicator label="Tasks Running">{running}</Metrics.Indicator>
-        <Metrics.Indicator
-          label="Tasks Failed"
-          isAlert
-          alertType={failed > 0 ? 'error' : 'success'}
+        </Metrics.Section>
+      </Metrics.Wrapper>
+
+      {showTraceModal && (
+        <Modal
+          isOpen={showTraceModal}
+          onClose={() => setShowTraceModal(false)}
+          title="Connector Error Details"
+          footer={
+            <Button
+              buttonType="primary"
+              buttonSize="M"
+              onClick={() => setShowTraceModal(false)}
+            >
+              Close
+            </Button>
+          }
         >
-          {failed}
-        </Metrics.Indicator>
-      </Metrics.Section>
-    </Metrics.Wrapper>
+          {connector.status.workerId && (
+            <S.WorkerInfo>Worker: {connector.status.workerId}</S.WorkerInfo>
+          )}
+
+          {connector.status.trace && (
+            <S.TraceContent>
+              <div>{connector.status.trace}</div>
+            </S.TraceContent>
+          )}
+        </Modal>
+      )}
+    </>
   );
 };
 
diff --git a/frontend/src/components/Connect/Details/Overview/__tests__/Overview.spec.tsx b/frontend/src/components/Connect/Details/Overview/__tests__/Overview.spec.tsx
diff --git a/frontend/src/components/common/Modal/Modal.styled.tsx b/frontend/src/components/common/Modal/Modal.styled.tsx
diff --git a/frontend/src/components/common/Modal/Modal.tsx b/frontend/src/components/common/Modal/Modal.tsx
diff --git a/frontend/src/components/common/Modal/index.ts b/frontend/src/components/common/Modal/index.ts
diff --git a/frontend/src/components/common/Tag/Tag.styled.tsx b/frontend/src/components/common/Tag/Tag.styled.tsx

Original file line number	Diff line number	Diff line change
`@@ -3,8 +3,10 @@`
`3`	`3`	`import io.kafbat.ui.model.KafkaCluster;`
`4`	`4`	`import reactor.core.publisher.Mono;`
`5`	`5`
`6`		`-public interface AdminClientService {`
	`6`	`+public interface AdminClientService extends AutoCloseable {`
`7`	`7`
`8`	`8`	`Mono<ReactiveAdminClient> get(KafkaCluster cluster);`
`9`	`9`
	`10`	`+ void invalidate(KafkaCluster cluster, Throwable e);`
	`11`	`+`
`10`	`12`	`}`
Original file line number	Diff line number	Diff line change
`@@ -73,7 +73,8 @@ public Flux<InternalBroker> getBrokers(KafkaCluster cluster) {`
`73`	`73`	`.map(description -> description.getNodes().stream()`
`74`	`74`	`.map(node -> new InternalBroker(node, partitionsDistribution, stats.getMetrics()))`
`75`	`75`	`.collect(Collectors.toList()))`
`76`		`- .flatMapMany(Flux::fromIterable);`
	`76`	`+ .flatMapMany(Flux::fromIterable)`
	`77`	`+ .doOnError(e -> adminClientService.invalidate(cluster, e));`
`77`	`78`	`}`
`78`	`79`
`79`	`80`	`public Mono<Void> updateBrokerLogDir(KafkaCluster cluster,`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,8 @@ private Mono<Statistics> getStatistics(KafkaCluster cluster) {`
`49`	`49`	`)`
`50`	`50`	`)`
`51`	`51`	`).doOnError(e ->`
`52`		`- log.error("Failed to collect cluster {} info", cluster.getName(), e))`
	`52`	`+ log.error("Failed to collect cluster {} info", cluster.getName(), e)`
	`53`	`+ ).doOnError(e -> adminClientService.invalidate(cluster, e))`
`53`	`54`	`.onErrorResume(t -> Mono.just(Statistics.statsUpdateError(t))));`
`54`	`55`	`}`
`55`	`56`
Original file line number	Diff line number	Diff line change
`@@ -217,6 +217,7 @@ enum ConnectorState {`
`217`	`217`	`model ConnectorStatus {`
`218`	`218`	`state: ConnectorState;`
`219`	`219`	`workerId?: string;`
	`220`	`+ trace?: string;`
`220`	`221`	`}`
`221`	`222`
`222`	`223`	`model Connector {`