topics connectors

Author: Leshe4ka

frontend/src/components/Connect/List/ConnectorsTable/ConnectorsTable.tsx

@@ -3,6 +3,7 @@ import { FullConnectorInfo } from 'generated-sources';
 import Table from 'components/common/NewTable';
 import { useLocalStoragePersister } from 'components/common/NewTable/ColumnResizer/lib';
 import { useQueryPersister } from 'components/common/NewTable/ColumnFilter';
+import { VisibilityState } from '@tanstack/react-table';
 
 import { connectorsColumns } from './connectorsColumns/columns';
 
@@ -12,11 +13,13 @@ const setRowId = (originalRow: FullConnectorInfo) =>
 type ConnectorsTableProps = {
   connectors: FullConnectorInfo[];
   columnSizingPersistKey?: string;
+  columnVisibility?: VisibilityState;
 };
 
 export const ConnectorsTable = ({
   connectors,
   columnSizingPersistKey = 'KafkaConnect',
+  columnVisibility,
 }: ConnectorsTableProps) => {
   const filterPersister = useQueryPersister(connectorsColumns);
   const columnSizingPersister = useLocalStoragePersister(
@@ -33,6 +36,7 @@ export const ConnectorsTable = ({
       emptyMessage="No connectors found"
       setRowId={setRowId}
       filterPersister={filterPersister}
+      columnVisibility={columnVisibility}
     />
   );
 };

frontend/src/components/Connect/List/ConnectorsTable/connectorsColumns/cells/ActionsCell.tsx

@@ -7,7 +7,7 @@ import {
   ResourceType,
 } from 'generated-sources';
 import { CellContext } from '@tanstack/react-table';
-import { ClusterNameRoute } from 'lib/paths';
+import { RouteParamsClusterTopic } from 'lib/paths';
 import useAppParams from 'lib/hooks/useAppParams';
 import { Dropdown } from 'components/common/Dropdown';
 import {
@@ -24,7 +24,7 @@ const ActionsCell: React.FC<CellContext<FullConnectorInfo, unknown>> = ({
   row,
 }) => {
   const { connect, name, status } = row.original;
-  const { clusterName } = useAppParams<ClusterNameRoute>();
+  const { clusterName, topicName } = useAppParams<RouteParamsClusterTopic>();
   const { isReadOnly } = useContext(ClusterContext);
   const mutationsNumber = useIsMutating();
   const isMutating = mutationsNumber > 0;
@@ -33,16 +33,19 @@ const ActionsCell: React.FC<CellContext<FullConnectorInfo, unknown>> = ({
     clusterName,
     connectName: connect,
     connectorName: name,
+    topicName,
   });
   const stateMutation = useUpdateConnectorState({
     clusterName,
     connectName: connect,
     connectorName: name,
+    topicName,
   });
   const resetConnectorOffsetsMutation = useResetConnectorOffsets({
     clusterName,
     connectName: connect,
     connectorName: name,
+    topicName,
   });
   const handleDelete = () => {
     confirm(
@@ -54,7 +57,7 @@ const ActionsCell: React.FC<CellContext<FullConnectorInfo, unknown>> = ({
       }
     );
   };
-  // const stateMutation = useUpdateConnectorState(routerProps);
+
   const resumeConnectorHandler = () =>
     stateMutation.mutateAsync(ConnectorAction.RESUME);
   const restartConnectorHandler = () =>

frontend/src/components/Topics/Topic/Connectors/Connectors.tsx

@@ -0,0 +1,17 @@
+import React from 'react';
+import { FullConnectorInfo } from 'generated-sources';
+import { ConnectorsTable } from 'components/Connect/List/ConnectorsTable/ConnectorsTable';
+
+type ConnectorsProps = {
+  connectors: FullConnectorInfo[];
+};
+
+const Connectors: React.FC<ConnectorsProps> = ({ connectors }) => (
+  <ConnectorsTable
+    columnVisibility={{ topics: false }}
+    connectors={connectors}
+    columnSizingPersistKey="topic-connectors"
+  />
+);
+
+export default Connectors;

frontend/src/components/Topics/Topic/Topic.tsx

@@ -1,6 +1,7 @@
 import React, { Suspense } from 'react';
 import { NavLink, Route, Routes, useNavigate } from 'react-router-dom';
 import {
+  clusterTopicConnectorsRelativePath,
   clusterTopicConsumerGroupsRelativePath,
   clusterTopicEditRelativePath,
   clusterTopicMessagesRelativePath,
@@ -22,6 +23,7 @@ import {
   useClearTopicMessages,
   useDeleteTopic,
   useRecreateTopic,
+  useTopicConnectors,
   useTopicDetails,
 } from 'lib/hooks/api/topics';
 import {
@@ -43,6 +45,7 @@ import Settings from './Settings/Settings';
 import TopicConsumerGroups from './ConsumerGroups/TopicConsumerGroups';
 import Statistics from './Statistics/Statistics';
 import Edit from './Edit/Edit';
+import Connectors from './Connectors/Connectors';
 import SendMessage from './SendMessage/SendMessage';
 
 const Topic: React.FC = () => {
@@ -70,6 +73,10 @@ const Topic: React.FC = () => {
   const deleteTopic = useDeleteTopic(clusterName);
   const recreateTopic = useRecreateTopic({ clusterName, topicName });
   const { data } = useTopicDetails({ clusterName, topicName });
+  const { data: connectors = [] } = useTopicConnectors({
+    clusterName,
+    topicName,
+  });
 
   const { isReadOnly, isTopicDeletionAllowed } =
     React.useContext(ClusterContext);
@@ -84,6 +91,7 @@ const Topic: React.FC = () => {
     await clearMessages.mutateAsync(topicName);
   };
   const canCleanup = data?.cleanUpPolicy === CleanUpPolicy.DELETE;
+  const isConnectorsAvailable = connectors.length > 0;
 
   return (
     <>
@@ -225,6 +233,19 @@ const Topic: React.FC = () => {
         >
           Statistics
         </ActionNavLink>
+        {isConnectorsAvailable && (
+          <ActionNavLink
+            to={clusterTopicConnectorsRelativePath}
+            className={({ isActive }) => (isActive ? 'is-active' : '')}
+            permission={{
+              resource: ResourceType.TOPIC,
+              action: Action.ANALYSIS_VIEW,
+              value: topicName,
+            }}
+          >
+            Connectors
+          </ActionNavLink>
+        )}
       </Navbar>
       <TopicActionsProvider openSidebarWithMessage={openSidebarWithMessage}>
         <Suspense fallback={<PageLoader />}>
@@ -246,6 +267,12 @@ const Topic: React.FC = () => {
               path={clusterTopicStatisticsRelativePath}
               element={<Statistics />}
             />
+            {isConnectorsAvailable && (
+              <Route
+                path={clusterTopicConnectorsRelativePath}
+                element={<Connectors connectors={connectors} />}
+              />
+            )}
             <Route path={clusterTopicEditRelativePath} element={<Edit />} />
           </Routes>
         </Suspense>

frontend/src/components/Topics/Topic/__test__/Topic.spec.tsx

@@ -21,6 +21,7 @@ import {
   useDeleteTopic,
   useRecreateTopic,
   useTopicDetails,
+  useTopicConnectors,
 } from 'lib/hooks/api/topics';
 
 const mockNavigate = jest.fn();
@@ -33,6 +34,7 @@ jest.mock('lib/hooks/api/topics', () => ({
   useDeleteTopic: jest.fn(),
   useRecreateTopic: jest.fn(),
   useClearTopicMessages: jest.fn(),
+  useTopicConnectors: jest.fn(),
 }));
 
 const clearTopicMessages = jest.fn();
@@ -98,6 +100,9 @@ describe('Details', () => {
     (useClearTopicMessages as jest.Mock).mockImplementation(() => ({
       mutateAsync: clearTopicMessages,
     }));
+    (useTopicConnectors as jest.Mock).mockImplementation(() => ({
+      data: [],
+    }));
   });
   describe('Action Bar', () => {
     describe('when it has readonly flag', () => {
@@ -132,7 +137,7 @@ describe('Details', () => {
 
     describe('when clear messages modal is open', () => {
       beforeEach(async () => {
-        await renderComponent();
+        renderComponent();
         const confirmButton = screen.getAllByText('Clear messages')[0];
         await userEvent.click(confirmButton);
       });

frontend/src/components/common/NewTable/Table.tsx

@@ -6,6 +6,7 @@ import type {
   PaginationState,
   Row,
   SortingState,
+  VisibilityState,
 } from '@tanstack/react-table';
 import {
   flexRender,
@@ -61,6 +62,9 @@ export interface TableProps<TData> {
   filterPersister?: Persister;
   resetPaginationOnFilter?: boolean;
 
+  // Columns visibility
+  columnVisibility?: VisibilityState;
+
   // Placeholder for empty table
   emptyMessage?: React.ReactNode;
 
@@ -158,6 +162,7 @@ function Table<TData>({
   setRowId,
   filterPersister,
   resetPaginationOnFilter = true,
+  columnVisibility,
 }: TableProps<TData>) {
   const [searchParams, setSearchParams] = useSearchParams();
   const location = useLocation();
@@ -209,6 +214,7 @@ function Table<TData>({
       columnFilters: filterPersister?.getPrevState() ?? [],
       rowSelection,
       columnSizing: columnSizingPersister?.columnSizing ?? {},
+      columnVisibility,
     },
     getRowId: (originalRow, index) => {
       if (setRowId) {

frontend/src/lib/hooks/api/kafkaConnect.ts

@@ -3,14 +3,17 @@ import {
   Connector,
   ConnectorAction,
   NewConnector,
+  Topic,
 } from 'generated-sources';
 import { kafkaConnectApiClient as api } from 'lib/api';
 import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
 import { ClusterName } from 'lib/interfaces/cluster';
 import { showSuccessAlert } from 'lib/errorHandling';
+import { topicKeys } from 'lib/hooks/api/topics';
 
 interface UseConnectorProps {
   clusterName: ClusterName;
+  topicName?: Topic['name'];
   connectName: Connect['name'];
   connectorName: Connector['name'];
 }
@@ -102,6 +105,13 @@ export function useUpdateConnectorState(props: UseConnectorProps) {
         Promise.all([
           client.invalidateQueries(connectorsKey(props.clusterName)),
           client.invalidateQueries(connectorKey(props)),
+          props.topicName &&
+            client.invalidateQueries(
+              topicKeys.connectors({
+                clusterName: props.clusterName,
+                topicName: props.topicName,
+              })
+            ),
         ]),
     }
   );

frontend/src/lib/hooks/api/topics.ts

@@ -19,6 +19,8 @@ import {
   TopicCreation,
   TopicDetails,
   TopicUpdate,
+  GetTopicConnectorsRequest,
+  FullConnectorInfo,
 } from 'generated-sources';
 import { showServerError, showSuccessAlert } from 'lib/errorHandling';
 import { ClusterName } from 'lib/interfaces/cluster';
@@ -45,6 +47,8 @@ export const topicKeys = {
     [...topicKeys.details(props), 'consumerGroups'] as const,
   statistics: (props: GetTopicDetailsRequest) =>
     [...topicKeys.details(props), 'statistics'] as const,
+  connectors: (props: GetTopicConnectorsRequest) =>
+    [...topicKeys.details(props), 'connectors'] as const,
 };
 
 export function useTopics(props: GetTopicsRequest) {
@@ -81,6 +85,17 @@ export function useTopicConsumerGroups(props: GetTopicDetailsRequest) {
   );
 }
 
+export function useTopicConnectors(
+  props: GetTopicConnectorsRequest,
+  queryOptions?: UseQueryOptions<FullConnectorInfo[]>
+) {
+  return useQuery<FullConnectorInfo[]>(
+    topicKeys.connectors(props),
+    () => api.getTopicConnectors(props),
+    queryOptions
+  );
+}
+
 const topicReducer = (
   result: TopicFormFormattedParams,
   customParam: TopicConfig

frontend/src/lib/paths.ts

@@ -156,6 +156,7 @@ export const clusterTopicSettingsRelativePath = 'settings';
 export const clusterTopicMessagesRelativePath = 'messages';
 export const clusterTopicConsumerGroupsRelativePath = 'consumer-groups';
 export const clusterTopicStatisticsRelativePath = 'statistics';
+export const clusterTopicConnectorsRelativePath = 'connectors';
 export const clusterTopicEditRelativePath = 'edit';
 export const clusterTopicPath = (
   clusterName: ClusterName = RouteParams.clusterName,