Issue#517 connector task status bug (#747)

* kafka-ui-connectors.yaml fix

* adding tasks status check

* adding new Connector Status

* adding new Connector Status

* [issue-517] connector task status bug #747

Co-authored-by: marselakhmetov <makhmetov@provectus.com>
Co-authored-by: mbovtryuk <mbovtryuk@provectus.com>

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.provectus.kafka.ui.client.KafkaConnectClients;
 import com.provectus.kafka.ui.connect.model.ConnectorTopics;
+import com.provectus.kafka.ui.connect.model.TaskStatus;
 import com.provectus.kafka.ui.exception.ClusterNotFoundException;
 import com.provectus.kafka.ui.exception.ConnectNotFoundException;
 import com.provectus.kafka.ui.mapper.ClusterMapper;
@@ -13,6 +14,7 @@ import com.provectus.kafka.ui.model.Connector;
 import com.provectus.kafka.ui.model.ConnectorAction;
 import com.provectus.kafka.ui.model.ConnectorPlugin;
 import com.provectus.kafka.ui.model.ConnectorPluginConfigValidationResponse;
+import com.provectus.kafka.ui.model.ConnectorState;
 import com.provectus.kafka.ui.model.FullConnectorInfo;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.model.KafkaConnectCluster;
@@ -161,23 +163,32 @@ public class KafkaConnectService {
   public Mono<Connector> getConnector(String clusterName, String connectName,
                                       String connectorName) {
     return getConnectAddress(clusterName, connectName)
-        .flatMap(connect ->
-            KafkaConnectClients.withBaseUrl(connect).getConnector(connectorName)
-                .map(kafkaConnectMapper::fromClient)
-                .flatMap(connector ->
-                    KafkaConnectClients.withBaseUrl(connect).getConnectorStatus(connector.getName())
-                        .map(connectorStatus -> {
-                          var status = connectorStatus.getConnector();
-                          connector.status(kafkaConnectMapper.fromClient(status));
-                          return (Connector) new Connector()
-                              .connect(connectName)
-                              .status(kafkaConnectMapper.fromClient(status))
-                              .type(connector.getType())
-                              .tasks(connector.getTasks())
-                              .name(connector.getName())
-                              .config(connector.getConfig());
-                        })
-                )
+        .flatMap(connect -> KafkaConnectClients.withBaseUrl(connect).getConnector(connectorName)
+            .map(kafkaConnectMapper::fromClient)
+            .flatMap(connector ->
+                KafkaConnectClients.withBaseUrl(connect).getConnectorStatus(connector.getName())
+                    .map(connectorStatus -> {
+                      var status = connectorStatus.getConnector();
+                      Connector result = (Connector) new Connector()
+                          .connect(connectName)
+                          .status(kafkaConnectMapper.fromClient(status))
+                          .type(connector.getType())
+                          .tasks(connector.getTasks())
+                          .name(connector.getName())
+                          .config(connector.getConfig());
+
+                      if (connectorStatus.getTasks() != null) {
+                        boolean isAnyTaskFailed = connectorStatus.getTasks().stream()
+                            .map(TaskStatus::getState)
+                            .anyMatch(TaskStatus.StateEnum.FAILED::equals);
+
+                        if (isAnyTaskFailed) {
+                          result.getStatus().state(ConnectorState.TASK_FAILED);
+                        }
+                      }
+                      return result;
+                    })
+            )
         );
   }
 

kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaConnectServiceTests.java

@@ -8,8 +8,8 @@ import com.provectus.kafka.ui.model.ConnectorPlugin;
 import com.provectus.kafka.ui.model.ConnectorPluginConfig;
 import com.provectus.kafka.ui.model.ConnectorPluginConfigValidationResponse;
 import com.provectus.kafka.ui.model.ConnectorPluginConfigValue;
+import com.provectus.kafka.ui.model.ConnectorState;
 import com.provectus.kafka.ui.model.ConnectorStatus;
-import com.provectus.kafka.ui.model.ConnectorTaskStatus;
 import com.provectus.kafka.ui.model.ConnectorType;
 import com.provectus.kafka.ui.model.NewConnector;
 import com.provectus.kafka.ui.model.TaskId;
@@ -171,7 +171,7 @@ public class KafkaConnectServiceTests extends AbstractBaseTest {
     Connector expected = (Connector) new Connector()
         .connect(connectName)
         .status(new ConnectorStatus()
-            .state(ConnectorTaskStatus.RUNNING)
+            .state(ConnectorState.RUNNING)
             .workerId("kafka-connect:8083"))
         .tasks(List.of(new TaskId()
             .connector(connectorName)

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -2285,7 +2285,7 @@ components:
       type: object
       properties:
         state:
-          $ref: '#/components/schemas/ConnectorTaskStatus'
+          $ref: '#/components/schemas/ConnectorState'
         worker_id:
           type: string
       required:
@@ -2299,6 +2299,15 @@ components:
         - PAUSED
         - UNASSIGNED
 
+    ConnectorState:
+      type: string
+      enum:
+        - RUNNING
+        - FAILED
+        - PAUSED
+        - UNASSIGNED
+        - TASK_FAILED
+
     ConnectorAction:
       type: string
       enum:

kafka-ui-react-app/src/components/Connect/ConnectorStatusTag.tsx

@@ -0,0 +1,21 @@
+import cx from 'classnames';
+import { ConnectorState } from 'generated-sources';
+import React from 'react';
+
+export interface StatusTagProps {
+  status: ConnectorState;
+}
+
+const ConnectorStatusTag: React.FC<StatusTagProps> = ({ status }) => {
+  const classNames = cx('tag', {
+    'is-success': status === ConnectorState.RUNNING,
+    'is-light': status === ConnectorState.PAUSED,
+    'is-warning': status === ConnectorState.UNASSIGNED,
+    'is-danger':
+      status === ConnectorState.FAILED || status === ConnectorState.TASK_FAILED,
+  });
+
+  return <span className={classNames}>{status}</span>;
+};
+
+export default ConnectorStatusTag;

kafka-ui-react-app/src/components/Connect/Details/Actions/Actions.tsx

@@ -1,6 +1,6 @@
 import React from 'react';
 import { Link, useHistory, useParams } from 'react-router-dom';
-import { ConnectorTaskStatus } from 'generated-sources';
+import { ConnectorState } from 'generated-sources';
 import { ClusterName, ConnectName, ConnectorName } from 'redux/interfaces';
 import {
   clusterConnectConnectorEditPath,
@@ -21,7 +21,7 @@ export interface ActionsProps {
     connectorName: ConnectorName
   ): Promise<void>;
   isConnectorDeleting: boolean;
-  connectorStatus?: ConnectorTaskStatus;
+  connectorStatus?: ConnectorState;
   restartConnector(
     clusterName: ClusterName,
     connectName: ConnectName,
@@ -79,7 +79,7 @@ const Actions: React.FC<ActionsProps> = ({
 
   return (
     <div className="buttons">
-      {connectorStatus === ConnectorTaskStatus.RUNNING && (
+      {connectorStatus === ConnectorState.RUNNING && (
         <button
           type="button"
           className="button"
@@ -93,7 +93,7 @@ const Actions: React.FC<ActionsProps> = ({
         </button>
       )}
 
-      {connectorStatus === ConnectorTaskStatus.PAUSED && (
+      {connectorStatus === ConnectorState.PAUSED && (
         <button
           type="button"
           className="button"

kafka-ui-react-app/src/components/Connect/Details/Actions/__tests__/Actions.spec.tsx

@@ -8,7 +8,7 @@ import ActionsContainer from 'components/Connect/Details/Actions/ActionsContaine
 import Actions, {
   ActionsProps,
 } from 'components/Connect/Details/Actions/Actions';
-import { ConnectorTaskStatus } from 'generated-sources';
+import { ConnectorState } from 'generated-sources';
 import { ConfirmationModalProps } from 'components/common/ConfirmationModal/ConfirmationModal';
 
 const mockHistoryPush = jest.fn();
@@ -45,7 +45,7 @@ describe('Actions', () => {
         <Actions
           deleteConnector={jest.fn()}
           isConnectorDeleting={false}
-          connectorStatus={ConnectorTaskStatus.RUNNING}
+          connectorStatus={ConnectorState.RUNNING}
           restartConnector={jest.fn()}
           pauseConnector={jest.fn()}
           resumeConnector={jest.fn()}
@@ -62,21 +62,21 @@ describe('Actions', () => {
 
     it('matches snapshot when paused', () => {
       const wrapper = create(
-        setupWrapper({ connectorStatus: ConnectorTaskStatus.PAUSED })
+        setupWrapper({ connectorStatus: ConnectorState.PAUSED })
       );
       expect(wrapper.toJSON()).toMatchSnapshot();
     });
 
     it('matches snapshot when failed', () => {
       const wrapper = create(
-        setupWrapper({ connectorStatus: ConnectorTaskStatus.FAILED })
+        setupWrapper({ connectorStatus: ConnectorState.FAILED })
       );
       expect(wrapper.toJSON()).toMatchSnapshot();
     });
 
     it('matches snapshot when unassigned', () => {
       const wrapper = create(
-        setupWrapper({ connectorStatus: ConnectorTaskStatus.UNASSIGNED })
+        setupWrapper({ connectorStatus: ConnectorState.UNASSIGNED })
       );
       expect(wrapper.toJSON()).toMatchSnapshot();
     });
@@ -157,7 +157,7 @@ describe('Actions', () => {
       const pauseConnector = jest.fn();
       const wrapper = mount(
         setupWrapper({
-          connectorStatus: ConnectorTaskStatus.RUNNING,
+          connectorStatus: ConnectorState.RUNNING,
           pauseConnector,
         })
       );
@@ -174,7 +174,7 @@ describe('Actions', () => {
       const resumeConnector = jest.fn();
       const wrapper = mount(
         setupWrapper({
-          connectorStatus: ConnectorTaskStatus.PAUSED,
+          connectorStatus: ConnectorState.PAUSED,
           resumeConnector,
         })
       );

kafka-ui-react-app/src/components/Connect/Details/Overview/Overview.tsx

@@ -1,6 +1,6 @@
 import React from 'react';
 import { Connector } from 'generated-sources';
-import StatusTag from 'components/Connect/StatusTag';
+import ConnectorStatusTag from 'components/Connect/ConnectorStatusTag';
 
 export interface OverviewProps {
   connector: Connector | null;
@@ -38,7 +38,7 @@ const Overview: React.FC<OverviewProps> = ({
           <tr>
             <th>State</th>
             <td>
-              <StatusTag status={connector.status.state} />
+              <ConnectorStatusTag status={connector.status.state} />
             </td>
           </tr>
           <tr>

kafka-ui-react-app/src/components/Connect/Details/Overview/__tests__/__snapshots__/Overview.spec.tsx.snap

@@ -37,9 +37,11 @@ exports[`Overview view matches snapshot 1`] = `
           State
         </th>
         <td>
-          <mock-StatusTag
-            status="RUNNING"
-          />
+          <span
+            className="tag is-success"
+          >
+            RUNNING
+          </span>
         </td>
       </tr>
       <tr>

kafka-ui-react-app/src/components/Connect/List/ListItem.tsx

@@ -10,7 +10,7 @@ import Dropdown from 'components/common/Dropdown/Dropdown';
 import DropdownDivider from 'components/common/Dropdown/DropdownDivider';
 import DropdownItem from 'components/common/Dropdown/DropdownItem';
 import ConfirmationModal from 'components/common/ConfirmationModal/ConfirmationModal';
-import StatusTag from 'components/Connect/StatusTag';
+import ConnectorStatusTag from 'components/Connect/ConnectorStatusTag';
 
 export interface ListItemProps {
   clusterName: ClusterName;
@@ -72,7 +72,7 @@ const ListItem: React.FC<ListItemProps> = ({
           ))}
         </div>
       </td>
-      <td>{status && <StatusTag status={status.state} />}</td>
+      <td>{status && <ConnectorStatusTag status={status.state} />}</td>
       <td>
         {runningTasks && (
           <span

kafka-ui-react-app/src/components/Connect/List/__tests__/__snapshots__/ListItem.spec.tsx.snap

@@ -132,7 +132,7 @@ exports[`Connectors ListItem matches snapshot 1`] = `
                 </div>
               </td>
               <td>
-                <StatusTag
+                <ConnectorStatusTag
                   status="RUNNING"
                 >
                   <span
@@ -140,7 +140,7 @@ exports[`Connectors ListItem matches snapshot 1`] = `
                   >
                     RUNNING
                   </span>
-                </StatusTag>
+                </ConnectorStatusTag>
               </td>
               <td>
                 <span

kafka-ui-react-app/src/redux/reducers/connect/__test__/fixtures.ts

@@ -1,6 +1,7 @@
 import {
   Connect,
   Connector,
+  ConnectorState,
   ConnectorTaskStatus,
   ConnectorType,
   FullConnectorInfo,
@@ -49,7 +50,7 @@ export const connectors: FullConnectorInfo[] = [
     type: ConnectorType.SOURCE,
     topics: ['test-topic'],
     status: {
-      state: ConnectorTaskStatus.RUNNING,
+      state: ConnectorState.RUNNING,
     },
     tasksCount: 2,
     failedTasksCount: 0,
@@ -61,7 +62,7 @@ export const connectors: FullConnectorInfo[] = [
     type: ConnectorType.SINK,
     topics: ['test-topic'],
     status: {
-      state: ConnectorTaskStatus.FAILED,
+      state: ConnectorState.FAILED,
     },
     tasksCount: 3,
     failedTasksCount: 1,
@@ -90,7 +91,7 @@ export const connector: Connector = {
   name: 'hdfs-source-connector',
   type: ConnectorType.SOURCE,
   status: {
-    state: ConnectorTaskStatus.RUNNING,
+    state: ConnectorState.RUNNING,
     workerId: 'kafka-connect0:8083',
   },
   config: {

kafka-ui-react-app/src/redux/reducers/connect/__test__/reducer.spec.ts

@@ -1,4 +1,4 @@
-import { ConnectorTaskStatus } from 'generated-sources';
+import { ConnectorState, ConnectorTaskStatus } from 'generated-sources';
 import {
   fetchConnectorsAction,
   fetchConnectorAction,
@@ -23,7 +23,7 @@ const runningConnectorState = {
       ...connector,
       status: {
         ...connector.status,
-        state: ConnectorTaskStatus.RUNNING,
+        state: ConnectorState.RUNNING,
       },
     },
     tasks: tasks.map((task) => ({
@@ -44,7 +44,7 @@ const pausedConnectorState = {
       ...connector,
       status: {
         ...connector.status,
-        state: ConnectorTaskStatus.PAUSED,
+        state: ConnectorState.PAUSED,
       },
     },
     tasks: tasks.map((task) => ({

kafka-ui-react-app/src/redux/reducers/connect/reducer.ts

@@ -2,7 +2,7 @@ import { getType } from 'typesafe-actions';
 import * as actions from 'redux/actions';
 import { ConnectState } from 'redux/interfaces/connect';
 import { Action } from 'redux/interfaces';
-import { ConnectorTaskStatus } from 'generated-sources';
+import { ConnectorState, ConnectorTaskStatus } from 'generated-sources';
 
 export const initialState: ConnectState = {
   connects: [],
@@ -53,7 +53,7 @@ const reducer = (state = initialState, action: Action): ConnectState => {
                 ...state.currentConnector.connector,
                 status: {
                   ...state.currentConnector.connector?.status,
-                  state: ConnectorTaskStatus.PAUSED,
+                  state: ConnectorState.PAUSED,
                 },
               }
             : null,
@@ -76,7 +76,7 @@ const reducer = (state = initialState, action: Action): ConnectState => {
                 ...state.currentConnector.connector,
                 status: {
                   ...state.currentConnector.connector?.status,
-                  state: ConnectorTaskStatus.RUNNING,
+                  state: ConnectorState.RUNNING,
                 },
               }
             : null,