0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Transform redux connect reducer and actions into toolkit slice. (#1883)

Browse source 
* integrate redux toolkit for connects

* uncomment test case

* remove unnecessary comment

* reduce duplication
This commit is contained in:
Arsen Simonyan 2022-04-27 13:37:25 +04:00 • committed by Arsen Simonyan
parent 521ba0cb2f
commit 3d04007883
36 changed files with 1409 additions and 1515 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -22,34 +22,34 @@ const ConnectorActionsWrapperStyled = styled.div`
`;
export interface ActionsProps {
deleteConnector(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): Promise<void>;
deleteConnector(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): Promise<unknown>;
isConnectorDeleting: boolean;
connectorStatus?: ConnectorState;
restartConnector(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): void;
restartTasks(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
action: ConnectorAction
): void;
pauseConnector(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): void;
resumeConnector(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): void;
restartConnector(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): void;
restartTasks(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
action: ConnectorAction;
}): void;
pauseConnector(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): void;
resumeConnector(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): void;
isConnectorActionRunning: boolean;
}
@ -73,7 +73,7 @@ const Actions: React.FC<ActionsProps> = ({
const deleteConnectorHandler = React.useCallback(async () => {
try {
await deleteConnector(clusterName, connectName, connectorName);
await deleteConnector({ clusterName, connectName, connectorName });
history.push(clusterConnectorsPath(clusterName));
} catch {
// do not redirect
@ -81,22 +81,27 @@ const Actions: React.FC<ActionsProps> = ({
}, [deleteConnector, clusterName, connectName, connectorName, history]);
const restartConnectorHandler = React.useCallback(() => {
restartConnector(clusterName, connectName, connectorName);
restartConnector({ clusterName, connectName, connectorName });
}, [restartConnector, clusterName, connectName, connectorName]);
const restartTasksHandler = React.useCallback(
(actionType) => {
restartTasks(clusterName, connectName, connectorName, actionType);
restartTasks({
clusterName,
connectName,
connectorName,
action: actionType,
});
},
[restartTasks, clusterName, connectName, connectorName]
);
const pauseConnectorHandler = React.useCallback(() => {
pauseConnector(clusterName, connectName, connectorName);
pauseConnector({ clusterName, connectName, connectorName });
}, [pauseConnector, clusterName, connectName, connectorName]);
const resumeConnectorHandler = React.useCallback(() => {
resumeConnector(clusterName, connectName, connectorName);
resumeConnector({ clusterName, connectName, connectorName });
}, [resumeConnector, clusterName, connectName, connectorName]);
return (

2
kafka-ui-react-app/src/components/Connect/Details/Actions/ActionsContainer.ts
View file

@ -7,7 +7,7 @@ import {
restartTasks,
pauseConnector,
resumeConnector,
} from 'redux/actions';
} from 'redux/reducers/connect/connectSlice';
import {
getIsConnectorDeleting,
getConnectorStatus,

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

@ -121,11 +121,11 @@ describe('Actions', () => {
wrapper.find('mock-ConfirmationModal').props() as ConfirmationModalProps
).onConfirm();
expect(deleteConnector).toHaveBeenCalledTimes(1);
expect(deleteConnector).toHaveBeenCalledWith(
expect(deleteConnector).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName
);
connectorName,
});
});
it('redirects after delete', async () => {
@ -151,11 +151,11 @@ describe('Actions', () => {
const wrapper = mount(setupWrapper({ restartConnector }));
wrapper.find({ children: 'Restart Connector' }).simulate('click');
expect(restartConnector).toHaveBeenCalledTimes(1);
expect(restartConnector).toHaveBeenCalledWith(
expect(restartConnector).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName
);
connectorName,
});
});
it('calls pauseConnector when pause button clicked', () => {
@ -168,11 +168,11 @@ describe('Actions', () => {
);
wrapper.find({ children: 'Pause' }).simulate('click');
expect(pauseConnector).toHaveBeenCalledTimes(1);
expect(pauseConnector).toHaveBeenCalledWith(
expect(pauseConnector).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName
);
connectorName,
});
});
it('calls resumeConnector when resume button clicked', () => {
@ -185,11 +185,11 @@ describe('Actions', () => {
);
wrapper.find({ children: 'Resume' }).simulate('click');
expect(resumeConnector).toHaveBeenCalledTimes(1);
expect(resumeConnector).toHaveBeenCalledWith(
expect(resumeConnector).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName
);
connectorName,
});
});
});
});

13
kafka-ui-react-app/src/components/Connect/Details/Config/Config.tsx
View file

@ -17,12 +17,11 @@ interface RouterParams {
}
export interface ConfigProps {
fetchConfig(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
silent?: boolean
): void;
fetchConfig(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): void;
isConfigFetching: boolean;
config: ConnectorConfig | null;
}
@ -39,7 +38,7 @@ const Config: React.FC<ConfigProps> = ({
const { clusterName, connectName, connectorName } = useParams<RouterParams>();
React.useEffect(() => {
fetchConfig(clusterName, connectName, connectorName, true);
fetchConfig({ clusterName, connectName, connectorName });
}, [fetchConfig, clusterName, connectName, connectorName]);
if (isConfigFetching) {

2
kafka-ui-react-app/src/components/Connect/Details/Config/ConfigContainer.ts
View file

@ -1,7 +1,7 @@
import { connect } from 'react-redux';
import { withRouter } from 'react-router-dom';
import { RootState } from 'redux/interfaces';
import { fetchConnectorConfig } from 'redux/actions';
import { fetchConnectorConfig } from 'redux/reducers/connect/connectSlice';
import {
getIsConnectorConfigFetching,
getConnectorConfig,

5
kafka-ui-react-app/src/components/Connect/Details/Config/__test__/Config.spec.tsx
View file

@ -61,12 +61,11 @@ describe('Config', () => {
const fetchConfig = jest.fn();
mount(setupWrapper({ fetchConfig }));
expect(fetchConfig).toHaveBeenCalledTimes(1);
expect(fetchConfig).toHaveBeenCalledWith(
expect(fetchConfig).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName,
true
);
});
});
});
});

24
kafka-ui-react-app/src/components/Connect/Details/Details.tsx
View file

@ -23,16 +23,16 @@ interface RouterParams {
}
export interface DetailsProps {
fetchConnector(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): void;
fetchTasks(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): void;
fetchConnector(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): void;
fetchTasks(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): void;
isConnectorFetching: boolean;
areTasksFetching: boolean;
connector: Connector | null;
@ -49,11 +49,11 @@ const Details: React.FC<DetailsProps> = ({
const { clusterName, connectName, connectorName } = useParams<RouterParams>();
React.useEffect(() => {
fetchConnector(clusterName, connectName, connectorName);
fetchConnector({ clusterName, connectName, connectorName });
}, [fetchConnector, clusterName, connectName, connectorName]);
React.useEffect(() => {
fetchTasks(clusterName, connectName, connectorName);
fetchTasks({ clusterName, connectName, connectorName });
}, [fetchTasks, clusterName, connectName, connectorName]);
if (isConnectorFetching || areTasksFetching) {

5
kafka-ui-react-app/src/components/Connect/Details/DetailsContainer.ts
View file

@ -1,7 +1,10 @@
import { connect } from 'react-redux';
import { withRouter } from 'react-router-dom';
import { RootState } from 'redux/interfaces';
import { fetchConnector, fetchConnectorTasks } from 'redux/actions';
import {
fetchConnector,
fetchConnectorTasks,
} from 'redux/reducers/connect/connectSlice';
import {
getIsConnectorFetching,
getAreConnectorTasksFetching,

19
kafka-ui-react-app/src/components/Connect/Details/Tasks/ListItem/ListItem.tsx
View file

@ -16,19 +16,24 @@ interface RouterParams {
export interface ListItemProps {
task: Task;
restartTask(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
taskId: TaskId['task']
): Promise<void>;
restartTask(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
taskId: TaskId['task'];
}): Promise<unknown>;
}
const ListItem: React.FC<ListItemProps> = ({ task, restartTask }) => {
const { clusterName, connectName, connectorName } = useParams<RouterParams>();
const restartTaskHandler = React.useCallback(async () => {
await restartTask(clusterName, connectName, connectorName, task.id?.task);
await restartTask({
clusterName,
connectName,
connectorName,
taskId: task.id?.task,
});
}, [restartTask, clusterName, connectName, connectorName, task.id?.task]);
return (

2
kafka-ui-react-app/src/components/Connect/Details/Tasks/ListItem/ListItemContainer.ts
View file

@ -2,7 +2,7 @@ import { connect } from 'react-redux';
import { RouteComponentProps, withRouter } from 'react-router-dom';
import { Task } from 'generated-sources';
import { RootState } from 'redux/interfaces';
import { restartConnectorTask } from 'redux/actions';
import { restartConnectorTask } from 'redux/reducers/connect/connectSlice';
import ListItem from './ListItem';

6
kafka-ui-react-app/src/components/Connect/Details/Tasks/ListItem/__tests__/ListItem.spec.tsx
View file

@ -63,11 +63,11 @@ describe('ListItem', () => {
userEvent.click(screen.getByRole('button'));
userEvent.click(screen.getByRole('menuitem'));
expect(restartTask).toBeCalledTimes(1);
expect(restartTask).toHaveBeenCalledWith(
expect(restartTask).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName,
task.id?.task
);
taskId: task.id?.task,
});
});
});

13
kafka-ui-react-app/src/components/Connect/Details/Tasks/Tasks.tsx
View file

@ -15,12 +15,11 @@ interface RouterParams {
}
export interface TasksProps {
fetchTasks(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
silent?: boolean
): void;
fetchTasks(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): void;
areTasksFetching: boolean;
tasks: Task[];
}
@ -33,7 +32,7 @@ const Tasks: React.FC<TasksProps> = ({
const { clusterName, connectName, connectorName } = useParams<RouterParams>();
React.useEffect(() => {
fetchTasks(clusterName, connectName, connectorName, true);
fetchTasks({ clusterName, connectName, connectorName });
}, [fetchTasks, clusterName, connectName, connectorName]);
if (areTasksFetching) {

2
kafka-ui-react-app/src/components/Connect/Details/Tasks/TasksContainer.ts
View file

@ -1,7 +1,7 @@
import { connect } from 'react-redux';
import { withRouter } from 'react-router-dom';
import { RootState } from 'redux/interfaces';
import { fetchConnectorTasks } from 'redux/actions';
import { fetchConnectorTasks } from 'redux/reducers/connect/connectSlice';
import {
getAreConnectorTasksFetching,
getConnectorTasks,

5
kafka-ui-react-app/src/components/Connect/Details/Tasks/__tests__/Tasks.spec.tsx
View file

@ -64,12 +64,11 @@ describe('Tasks', () => {
const fetchTasks = jest.fn();
mount(setupWrapper({ fetchTasks }));
expect(fetchTasks).toHaveBeenCalledTimes(1);
expect(fetchTasks).toHaveBeenCalledWith(
expect(fetchTasks).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName,
true
);
});
});
});
});

12
kafka-ui-react-app/src/components/Connect/Details/__tests__/Details.spec.tsx
View file

@ -90,22 +90,22 @@ describe('Details', () => {
const fetchConnector = jest.fn();
render(setupWrapper({ fetchConnector }));
expect(fetchConnector).toHaveBeenCalledTimes(1);
expect(fetchConnector).toHaveBeenCalledWith(
expect(fetchConnector).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName
);
connectorName,
});
});
it('fetches tasks on mount', () => {
const fetchTasks = jest.fn();
render(setupWrapper({ fetchTasks }));
expect(fetchTasks).toHaveBeenCalledTimes(1);
expect(fetchTasks).toHaveBeenCalledWith(
expect(fetchTasks).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName
);
connectorName,
});
});
});
});

31
kafka-ui-react-app/src/components/Connect/Edit/Edit.tsx
View file

@ -3,7 +3,6 @@ import { useHistory, useParams } from 'react-router-dom';
import { Controller, useForm } from 'react-hook-form';
import { ErrorMessage } from '@hookform/error-message';
import { yupResolver } from '@hookform/resolvers/yup';
import { Connector } from 'generated-sources';
import {
ClusterName,
ConnectName,
@ -36,19 +35,19 @@ interface FormValues {
}
export interface EditProps {
fetchConfig(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): Promise<void>;
fetchConfig(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}): Promise<unknown>;
isConfigFetching: boolean;
config: ConnectorConfig | null;
updateConfig(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
connectorConfig: ConnectorConfig
): Promise<Connector | undefined>;
updateConfig(payload: {
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
connectorConfig: ConnectorConfig;
}): Promise<unknown>;
}
const Edit: React.FC<EditProps> = ({
@ -73,7 +72,7 @@ const Edit: React.FC<EditProps> = ({
});
React.useEffect(() => {
fetchConfig(clusterName, connectName, connectorName);
fetchConfig({ clusterName, connectName, connectorName });
}, [fetchConfig, clusterName, connectName, connectorName]);
React.useEffect(() => {
@ -84,12 +83,12 @@ const Edit: React.FC<EditProps> = ({
const onSubmit = React.useCallback(
async (values: FormValues) => {
const connector = await updateConfig(
const connector = await updateConfig({
clusterName,
connectName,
connectorName,
JSON.parse(values.config.trim())
);
connectorConfig: JSON.parse(values.config.trim()),
});
if (connector) {
history.push(
clusterConnectConnectorConfigPath(

5
kafka-ui-react-app/src/components/Connect/Edit/EditContainer.ts
View file

@ -1,7 +1,10 @@
import { connect } from 'react-redux';
import { withRouter } from 'react-router-dom';
import { RootState } from 'redux/interfaces';
import { fetchConnectorConfig, updateConnectorConfig } from 'redux/actions';
import {
fetchConnectorConfig,
updateConnectorConfig,
} from 'redux/reducers/connect/connectSlice';
import {
getConnectorConfig,
getIsConnectorConfigFetching,

12
kafka-ui-react-app/src/components/Connect/Edit/__tests__/Edit.spec.tsx
View file

@ -60,11 +60,11 @@ describe('Edit', () => {
const fetchConfig = jest.fn();
renderComponent({ fetchConfig });
expect(fetchConfig).toHaveBeenCalledTimes(1);
expect(fetchConfig).toHaveBeenCalledWith(
expect(fetchConfig).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName
);
connectorName,
});
});
it('calls updateConfig on form submit', async () => {
@ -72,12 +72,12 @@ describe('Edit', () => {
renderComponent({ updateConfig });
await waitFor(() => fireEvent.submit(screen.getByRole('form')));
expect(updateConfig).toHaveBeenCalledTimes(1);
expect(updateConfig).toHaveBeenCalledWith(
expect(updateConfig).toHaveBeenCalledWith({
clusterName,
connectName,
connectorName,
connector.config
);
connectorConfig: connector.config,
});
});
it('redirects to connector config view on successful submit', async () => {

4
kafka-ui-react-app/src/components/Connect/List/List.tsx
View file

@ -21,7 +21,7 @@ export interface ListProps {
connectors: FullConnectorInfo[];
connects: Connect[];
fetchConnects(clusterName: ClusterName): void;
fetchConnectors(clusterName: ClusterName): void;
fetchConnectors({ clusterName }: { clusterName: ClusterName }): void;
search: string;
setConnectorSearch(value: ConnectorSearch): void;
}
@ -40,7 +40,7 @@ const List: React.FC<ListProps> = ({
React.useEffect(() => {
fetchConnects(clusterName);
fetchConnectors(clusterName);
fetchConnectors({ clusterName });
}, [fetchConnects, fetchConnectors, clusterName]);
const handleSearch = (value: string) =>

2
kafka-ui-react-app/src/components/Connect/List/ListContainer.ts
View file

@ -4,7 +4,7 @@ import {
fetchConnects,
fetchConnectors,
setConnectorSearch,
} from 'redux/actions/thunks/connectors';
} from 'redux/reducers/connect/connectSlice';
import {
getConnects,
getConnectors,

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

@ -4,7 +4,7 @@ import { clusterConnectConnectorPath, clusterTopicPath } from 'lib/paths';
import { ClusterName } from 'redux/interfaces';
import { Link, NavLink } from 'react-router-dom';
import { useDispatch } from 'react-redux';
import { deleteConnector } from 'redux/actions';
import { deleteConnector } from 'redux/reducers/connect/connectSlice';
import Dropdown from 'components/common/Dropdown/Dropdown';
import DropdownItem from 'components/common/Dropdown/DropdownItem';
import ConfirmationModal from 'components/common/ConfirmationModal/ConfirmationModal';
@ -41,7 +41,13 @@ const ListItem: React.FC<ListItemProps> = ({
const handleDelete = React.useCallback(() => {
if (clusterName && connect && name) {
dispatch(deleteConnector(clusterName, connect, name));
dispatch(
deleteConnector({
clusterName,
connectName: connect,
connectorName: name,
})
);
}
setDeleteConnectorConfirmationVisible(false);
}, [clusterName, connect, dispatch, name]);

4
kafka-ui-react-app/src/components/Connect/List/__tests__/ListItem.spec.tsx
View file

@ -11,8 +11,8 @@ import theme from 'theme/theme';
const mockDeleteConnector = jest.fn(() => ({ type: 'test' }));
jest.mock('redux/actions', () => ({
...jest.requireActual('redux/actions'),
jest.mock('redux/reducers/connect/connectSlice', () => ({
...jest.requireActual('redux/reducers/connect/connectSlice'),
deleteConnector: () => mockDeleteConnector,
}));

19
kafka-ui-react-app/src/components/Connect/New/New.tsx
View file

@ -28,14 +28,14 @@ interface RouterParams {
}
export interface NewProps {
fetchConnects(clusterName: ClusterName): void;
fetchConnects(clusterName: ClusterName): unknown;
areConnectsFetching: boolean;
connects: Connect[];
createConnector(
clusterName: ClusterName,
connectName: ConnectName,
newConnector: NewConnector
): Promise<Connector | undefined>;
createConnector(payload: {
clusterName: ClusterName;
connectName: ConnectName;
newConnector: NewConnector;
}): Promise<{ connector: Connector | undefined }>;
}
interface FormValues {
@ -87,10 +87,15 @@ const New: React.FC<NewProps> = ({
const onSubmit = React.useCallback(
async (values: FormValues) => {
const connector = await createConnector(clusterName, values.connectName, {
const { connector } = await createConnector({
clusterName,
connectName: values.connectName,
newConnector: {
name: values.name,
config: JSON.parse(values.config.trim()),
},
});
if (connector) {
history.push(
clusterConnectConnectorPath(

9
kafka-ui-react-app/src/components/Connect/New/NewContainer.ts
View file

@ -1,13 +1,16 @@
import { connect } from 'react-redux';
import { withRouter } from 'react-router-dom';
import { createConnector, fetchConnects } from 'redux/actions';
import {
createConnector,
fetchConnects,
} from 'redux/reducers/connect/connectSlice';
import { RootState } from 'redux/interfaces';
import {
getAreConnectsFetching,
getConnects,
} from 'redux/reducers/connect/selectors';
import New from './New';
import New, { NewProps } from './New';
const mapStateToProps = (state: RootState) => ({
areConnectsFetching: getAreConnectsFetching(state),
@ -16,7 +19,7 @@ const mapStateToProps = (state: RootState) => ({
const mapDispatchToProps = {
fetchConnects,
createConnector,
createConnector: createConnector as unknown as NewProps['createConnector'],
};
export default withRouter(connect(mapStateToProps, mapDispatchToProps)(New));

12
kafka-ui-react-app/src/components/Connect/New/__tests__/New.spec.tsx
View file

@ -79,18 +79,18 @@ describe('New', () => {
renderComponent({ createConnector });
await simulateFormSubmit();
expect(createConnector).toHaveBeenCalledTimes(1);
expect(createConnector).toHaveBeenCalledWith(
expect(createConnector).toHaveBeenCalledWith({
clusterName,
connects[0].name,
{
connectName: connects[0].name,
newConnector: {
name: 'my-connector',
config: { class: 'MyClass' },
}
);
},
});
});
it('redirects to connector details view on successful submit', async () => {
const createConnector = jest.fn().mockResolvedValue(connector);
const createConnector = jest.fn().mockResolvedValue({ connector });
renderComponent({ createConnector });
await simulateFormSubmit();
expect(mockHistoryPush).toHaveBeenCalledTimes(1);

5
kafka-ui-react-app/src/lib/testHelpers.tsx
View file

@ -11,6 +11,7 @@ import { AnyAction, Store } from 'redux';
import { RootState } from 'redux/interfaces';
import { configureStore } from '@reduxjs/toolkit';
import rootReducer from 'redux/reducers';
import mockStoreCreator from 'redux/store/configureStore/mockStoreCreator';
interface TestRouterWrapperProps {
pathname: string;
@ -121,3 +122,7 @@ export class EventSourceMock {
this.close = jest.fn();
}
}
export const getTypeAndPayload = (store: typeof mockStoreCreator) => {
return store.getActions().map(({ type, payload }) => ({ type, payload }));
};

563
kafka-ui-react-app/src/redux/actions/__test__/thunks/connectors.spec.ts
View file

@ -1,563 +0,0 @@
import fetchMock from 'fetch-mock-jest';
import { ConnectorAction } from 'generated-sources';
import * as actions from 'redux/actions/actions';
import * as thunks from 'redux/actions/thunks';
import {
connects,
connectorsServerPayload,
connectors,
connectorServerPayload,
connector,
tasksServerPayload,
tasks,
} from 'redux/reducers/connect/__test__/fixtures';
import mockStoreCreator from 'redux/store/configureStore/mockStoreCreator';
const store = mockStoreCreator;
const clusterName = 'local';
const connectName = 'first';
const connectorName = 'hdfs-source-connector';
const taskId = 10;
describe('Thunks', () => {
afterEach(() => {
fetchMock.restore();
store.clearActions();
});
describe('fetchConnects', () => {
it('creates GET_CONNECTS__SUCCESS when fetching connects', async () => {
fetchMock.getOnce(`/api/clusters/${clusterName}/connects`, connects);
await store.dispatch(thunks.fetchConnects(clusterName));
expect(store.getActions()).toEqual([
actions.fetchConnectsAction.request(),
actions.fetchConnectsAction.success({ connects }),
]);
});
it('creates GET_CONNECTS__FAILURE', async () => {
fetchMock.getOnce(`/api/clusters/${clusterName}/connects`, 404);
await store.dispatch(thunks.fetchConnects(clusterName));
expect(store.getActions()).toEqual([
actions.fetchConnectsAction.request(),
actions.fetchConnectsAction.failure({
alert: {
subject: 'connects',
title: 'Kafka Connect',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects`,
},
},
}),
]);
});
});
describe('fetchConnectors', () => {
it('creates GET_CONNECTORS__SUCCESS when fetching connectors', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connectors`,
connectorsServerPayload,
{ query: { search: '' } }
);
await store.dispatch(thunks.fetchConnectors(clusterName));
expect(store.getActions()).toEqual([
actions.fetchConnectorsAction.request(),
actions.fetchConnectorsAction.success({ connectors }),
]);
});
it('creates GET_CONNECTORS__SUCCESS when fetching connectors in silent mode', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connectors`,
connectorsServerPayload,
{ query: { search: '' } }
);
await store.dispatch(thunks.fetchConnectors(clusterName, '', true));
expect(store.getActions()).toEqual([
actions.fetchConnectorsAction.success({
...store.getState().connect,
connectors,
}),
]);
});
it('creates GET_CONNECTORS__FAILURE', async () => {
fetchMock.getOnce(`/api/clusters/${clusterName}/connectors`, 404, {
query: { search: '' },
});
await store.dispatch(thunks.fetchConnectors(clusterName));
expect(store.getActions()).toEqual([
actions.fetchConnectorsAction.request(),
actions.fetchConnectorsAction.failure({
alert: {
subject: 'local-connectors',
title: 'Kafka Connect Connectors',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connectors?search=`,
},
},
}),
]);
});
});
describe('fetchConnector', () => {
it('creates GET_CONNECTOR__SUCCESS when fetching connects', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
connectorServerPayload
);
await store.dispatch(
thunks.fetchConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.fetchConnectorAction.request(),
actions.fetchConnectorAction.success({ connector }),
]);
});
it('creates GET_CONNECTOR__FAILURE', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
404
);
await store.dispatch(
thunks.fetchConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.fetchConnectorAction.request(),
actions.fetchConnectorAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
},
},
}),
]);
});
});
describe('createConnector', () => {
it('creates POST_CONNECTOR__SUCCESS when fetching connects', async () => {
fetchMock.postOnce(
{
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors`,
body: {
name: connectorName,
config: connector.config,
},
},
connectorServerPayload
);
await store.dispatch(
thunks.createConnector(clusterName, connectName, {
name: connectorName,
config: connector.config,
})
);
expect(store.getActions()).toEqual([
actions.createConnectorAction.request(),
actions.createConnectorAction.success({ connector }),
]);
});
it('creates POST_CONNECTOR__FAILURE', async () => {
fetchMock.postOnce(
{
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors`,
body: {
name: connectorName,
config: connector.config,
},
},
404
);
await store.dispatch(
thunks.createConnector(clusterName, connectName, {
name: connectorName,
config: connector.config,
})
);
expect(store.getActions()).toEqual([
actions.createConnectorAction.request(),
actions.createConnectorAction.failure({
alert: {
subject: 'local-first',
title: `Connector with name ${connectorName} already exists`,
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors`,
},
},
}),
]);
});
});
describe('deleteConnector', () => {
it('creates DELETE_CONNECTOR__SUCCESS', async () => {
fetchMock.deleteOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
{}
);
fetchMock.getOnce(
`/api/clusters/${clusterName}/connectors?search=`,
connectorsServerPayload
);
await store.dispatch(
thunks.deleteConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.deleteConnectorAction.request(),
actions.deleteConnectorAction.success({ connectorName }),
]);
});
it('creates DELETE_CONNECTOR__FAILURE', async () => {
fetchMock.deleteOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
404
);
try {
await store.dispatch(
thunks.deleteConnector(clusterName, connectName, connectorName)
);
} catch {
expect(store.getActions()).toEqual([
actions.deleteConnectorAction.request(),
actions.deleteConnectorAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Delete',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
},
},
}),
]);
}
});
});
describe('fetchConnectorTasks', () => {
it('creates GET_CONNECTOR_TASKS__SUCCESS when fetching connects', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
tasksServerPayload
);
await store.dispatch(
thunks.fetchConnectorTasks(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.fetchConnectorTasksAction.request(),
actions.fetchConnectorTasksAction.success({ tasks }),
]);
});
it('creates GET_CONNECTOR_TASKS__FAILURE', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
404
);
await store.dispatch(
thunks.fetchConnectorTasks(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.fetchConnectorTasksAction.request(),
actions.fetchConnectorTasksAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Tasks',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
},
},
}),
]);
});
});
describe('restartConnector', () => {
it('creates RESTART_CONNECTOR__SUCCESS when fetching connects', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESTART}`,
{ message: 'success' }
);
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
tasksServerPayload
);
await store.dispatch(
thunks.restartConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.restartConnectorAction.request(),
actions.restartConnectorAction.success(),
]);
});
it('creates RESTART_CONNECTOR__FAILURE', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESTART}`,
404
);
await store.dispatch(
thunks.restartConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.restartConnectorAction.request(),
actions.restartConnectorAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Restart',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESTART}`,
},
},
}),
]);
});
});
describe('pauseConnector', () => {
it('creates PAUSE_CONNECTOR__SUCCESS when fetching connects', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.PAUSE}`,
{ message: 'success' }
);
await store.dispatch(
thunks.pauseConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.pauseConnectorAction.request(),
actions.pauseConnectorAction.success({ connectorName }),
]);
});
it('creates PAUSE_CONNECTOR__FAILURE', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.PAUSE}`,
404
);
await store.dispatch(
thunks.pauseConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.pauseConnectorAction.request(),
actions.pauseConnectorAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Pause',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.PAUSE}`,
},
},
}),
]);
});
});
describe('resumeConnector', () => {
it('creates RESUME_CONNECTOR__SUCCESS when fetching connects', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESUME}`,
{ message: 'success' }
);
await store.dispatch(
thunks.resumeConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.resumeConnectorAction.request(),
actions.resumeConnectorAction.success({ connectorName }),
]);
});
it('creates RESUME_CONNECTOR__FAILURE', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESUME}`,
404
);
await store.dispatch(
thunks.resumeConnector(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.resumeConnectorAction.request(),
actions.resumeConnectorAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Resume',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESUME}`,
},
},
}),
]);
});
});
describe('restartConnectorTask', () => {
it('creates RESTART_CONNECTOR_TASK__SUCCESS when fetching connects', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks/${taskId}/action/restart`,
{ message: 'success' }
);
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
tasksServerPayload
);
await store.dispatch(
thunks.restartConnectorTask(
clusterName,
connectName,
connectorName,
taskId
)
);
expect(store.getActions()).toEqual([
actions.restartConnectorTaskAction.request(),
actions.restartConnectorTaskAction.success(),
]);
});
it('creates RESTART_CONNECTOR_TASK__FAILURE', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks/${taskId}/action/restart`,
404
);
await store.dispatch(
thunks.restartConnectorTask(
clusterName,
connectName,
connectorName,
taskId
)
);
expect(store.getActions()).toEqual([
actions.restartConnectorTaskAction.request(),
actions.restartConnectorTaskAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector-10',
title: 'Kafka Connect Connector Task Restart',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks/${taskId}/action/restart`,
},
},
}),
]);
});
});
describe('fetchConnectorConfig', () => {
it('creates GET_CONNECTOR_CONFIG__SUCCESS when fetching connects', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
connector.config
);
await store.dispatch(
thunks.fetchConnectorConfig(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.fetchConnectorConfigAction.request(),
actions.fetchConnectorConfigAction.success({
config: connector.config,
}),
]);
});
it('creates GET_CONNECTOR_CONFIG__FAILURE', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
404
);
await store.dispatch(
thunks.fetchConnectorConfig(clusterName, connectName, connectorName)
);
expect(store.getActions()).toEqual([
actions.fetchConnectorConfigAction.request(),
actions.fetchConnectorConfigAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Config',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
},
},
}),
]);
});
});
describe('updateConnectorConfig', () => {
it('creates PATCH_CONNECTOR_CONFIG__SUCCESS when fetching connects', async () => {
fetchMock.putOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
connectorServerPayload
);
await store.dispatch(
thunks.updateConnectorConfig(
clusterName,
connectName,
connectorName,
connector.config
)
);
expect(store.getActions()).toEqual([
actions.updateConnectorConfigAction.request(),
actions.updateConnectorConfigAction.success({ connector }),
]);
});
it('creates PATCH_CONNECTOR_CONFIG__FAILURE', async () => {
fetchMock.putOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
404
);
await store.dispatch(
thunks.updateConnectorConfig(
clusterName,
connectName,
connectorName,
connector.config
)
);
expect(store.getActions()).toEqual([
actions.updateConnectorConfigAction.request(),
actions.updateConnectorConfigAction.failure({
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Config Update',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
},
},
}),
]);
});
});
});

90
kafka-ui-react-app/src/redux/actions/actions.ts
View file

@ -1,17 +1,7 @@
import { createAction, createAsyncAction } from 'typesafe-actions';
import {
FailurePayload,
TopicName,
TopicsState,
ConnectorName,
ConnectorConfig,
} from 'redux/interfaces';
import { FailurePayload, TopicName, TopicsState } from 'redux/interfaces';
import {
TopicColumnsToSort,
Connector,
FullConnectorInfo,
Connect,
Task,
Topic,
TopicMessage,
TopicMessageConsuming,
@ -70,84 +60,6 @@ export const recreateTopicAction = createAsyncAction(
export const dismissAlert = createAction('DISMISS_ALERT')<string>();
export const fetchConnectsAction = createAsyncAction(
'GET_CONNECTS__REQUEST',
'GET_CONNECTS__SUCCESS',
'GET_CONNECTS__FAILURE'
)<undefined, { connects: Connect[] }, { alert?: FailurePayload }>();
export const fetchConnectorsAction = createAsyncAction(
'GET_CONNECTORS__REQUEST',
'GET_CONNECTORS__SUCCESS',
'GET_CONNECTORS__FAILURE'
)<undefined, { connectors: FullConnectorInfo[] }, { alert?: FailurePayload }>();
export const fetchConnectorAction = createAsyncAction(
'GET_CONNECTOR__REQUEST',
'GET_CONNECTOR__SUCCESS',
'GET_CONNECTOR__FAILURE'
)<undefined, { connector: Connector }, { alert?: FailurePayload }>();
export const createConnectorAction = createAsyncAction(
'POST_CONNECTOR__REQUEST',
'POST_CONNECTOR__SUCCESS',
'POST_CONNECTOR__FAILURE'
)<undefined, { connector: Connector }, { alert?: FailurePayload }>();
export const deleteConnectorAction = createAsyncAction(
'DELETE_CONNECTOR__REQUEST',
'DELETE_CONNECTOR__SUCCESS',
'DELETE_CONNECTOR__FAILURE'
)<undefined, { connectorName: ConnectorName }, { alert?: FailurePayload }>();
export const restartConnectorAction = createAsyncAction(
'RESTART_CONNECTOR__REQUEST',
'RESTART_CONNECTOR__SUCCESS',
'RESTART_CONNECTOR__FAILURE'
)<undefined, undefined, { alert?: FailurePayload }>();
export const restartTasksAction = createAsyncAction(
'RESTART_TASKS__REQUEST',
'RESTART_TASKS__SUCCESS',
'RESTART_TASKS__FAILURE'
)<undefined, undefined, { alert?: FailurePayload }>();
export const pauseConnectorAction = createAsyncAction(
'PAUSE_CONNECTOR__REQUEST',
'PAUSE_CONNECTOR__SUCCESS',
'PAUSE_CONNECTOR__FAILURE'
)<undefined, { connectorName: ConnectorName }, { alert?: FailurePayload }>();
export const resumeConnectorAction = createAsyncAction(
'RESUME_CONNECTOR__REQUEST',
'RESUME_CONNECTOR__SUCCESS',
'RESUME_CONNECTOR__FAILURE'
)<undefined, { connectorName: ConnectorName }, { alert?: FailurePayload }>();
export const fetchConnectorTasksAction = createAsyncAction(
'GET_CONNECTOR_TASKS__REQUEST',
'GET_CONNECTOR_TASKS__SUCCESS',
'GET_CONNECTOR_TASKS__FAILURE'
)<undefined, { tasks: Task[] }, { alert?: FailurePayload }>();
export const restartConnectorTaskAction = createAsyncAction(
'RESTART_CONNECTOR_TASK__REQUEST',
'RESTART_CONNECTOR_TASK__SUCCESS',
'RESTART_CONNECTOR_TASK__FAILURE'
)<undefined, undefined, { alert?: FailurePayload }>();
export const fetchConnectorConfigAction = createAsyncAction(
'GET_CONNECTOR_CONFIG__REQUEST',
'GET_CONNECTOR_CONFIG__SUCCESS',
'GET_CONNECTOR_CONFIG__FAILURE'
)<undefined, { config: ConnectorConfig }, { alert?: FailurePayload }>();
export const updateConnectorConfigAction = createAsyncAction(
'PATCH_CONNECTOR_CONFIG__REQUEST',
'PATCH_CONNECTOR_CONFIG__SUCCESS',
'PATCH_CONNECTOR_CONFIG__FAILURE'
)<undefined, { connector: Connector }, { alert?: FailurePayload }>();
export const setTopicsSearchAction =
createAction('SET_TOPICS_SEARCH')<string>();

391
kafka-ui-react-app/src/redux/actions/thunks/connectors.ts
View file

@ -1,391 +0,0 @@
import {
KafkaConnectApi,
Configuration,
NewConnector,
Connector,
ConnectorAction,
TaskId,
} from 'generated-sources';
import { BASE_PARAMS } from 'lib/constants';
import {
ClusterName,
ConnectName,
ConnectorConfig,
ConnectorName,
ConnectorSearch,
FailurePayload,
PromiseThunkResult,
} from 'redux/interfaces';
import * as actions from 'redux/actions';
import { getResponse } from 'lib/errorHandling';
import { batch } from 'react-redux';
const apiClientConf = new Configuration(BASE_PARAMS);
export const kafkaConnectApiClient = new KafkaConnectApi(apiClientConf);
export const fetchConnects =
(clusterName: ClusterName): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.fetchConnectsAction.request());
try {
const connects = await kafkaConnectApiClient.getConnects({ clusterName });
dispatch(actions.fetchConnectsAction.success({ connects }));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: 'connects',
title: `Kafka Connect`,
response,
};
dispatch(actions.fetchConnectsAction.failure({ alert }));
}
};
export const fetchConnectors =
(
clusterName: ClusterName,
search = '',
silent = false
): PromiseThunkResult<void> =>
async (dispatch) => {
if (!silent) dispatch(actions.fetchConnectorsAction.request());
try {
const connectors = await kafkaConnectApiClient.getAllConnectors({
clusterName,
search,
});
dispatch(actions.fetchConnectorsAction.success({ connectors }));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, 'connectors'].join('-'),
title: `Kafka Connect Connectors`,
response,
};
dispatch(actions.fetchConnectorsAction.failure({ alert }));
}
};
export const fetchConnector =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.fetchConnectorAction.request());
try {
const connector = await kafkaConnectApiClient.getConnector({
clusterName,
connectName,
connectorName,
});
dispatch(actions.fetchConnectorAction.success({ connector }));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector`,
response,
};
dispatch(actions.fetchConnectorAction.failure({ alert }));
}
};
export const createConnector =
(
clusterName: ClusterName,
connectName: ConnectName,
newConnector: NewConnector
): PromiseThunkResult<Connector | undefined> =>
async (dispatch) => {
dispatch(actions.createConnectorAction.request());
try {
const connector = await kafkaConnectApiClient.createConnector({
clusterName,
connectName,
newConnector,
});
dispatch(actions.createConnectorAction.success({ connector }));
return connector;
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName].join('-'),
title: `Connector with name ${newConnector.name} already exists`,
response,
};
dispatch(actions.createConnectorAction.failure({ alert }));
}
return undefined;
};
export const deleteConnector =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.deleteConnectorAction.request());
try {
await kafkaConnectApiClient.deleteConnector({
clusterName,
connectName,
connectorName,
});
dispatch(actions.deleteConnectorAction.success({ connectorName }));
dispatch(fetchConnectors(clusterName, '', true));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Delete`,
response,
};
dispatch(actions.deleteConnectorAction.failure({ alert }));
throw error;
}
};
export const fetchConnectorTasks =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
silent = false
): PromiseThunkResult<void> =>
async (dispatch) => {
if (!silent) dispatch(actions.fetchConnectorTasksAction.request());
try {
const tasks = await kafkaConnectApiClient.getConnectorTasks({
clusterName,
connectName,
connectorName,
});
dispatch(actions.fetchConnectorTasksAction.success({ tasks }));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Tasks`,
response,
};
dispatch(actions.fetchConnectorTasksAction.failure({ alert }));
}
};
export const restartConnector =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.restartConnectorAction.request());
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action: ConnectorAction.RESTART,
});
dispatch(actions.restartConnectorAction.success());
dispatch(
fetchConnectorTasks(clusterName, connectName, connectorName, true)
);
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Restart`,
response,
};
dispatch(actions.restartConnectorAction.failure({ alert }));
}
};
export const restartTasks =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
action: ConnectorAction
): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.restartTasksAction.request());
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action,
});
batch(() => {
dispatch(actions.restartTasksAction.success());
dispatch(
fetchConnectorTasks(clusterName, connectName, connectorName, true)
);
});
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Tasks Restart`,
response,
};
dispatch(actions.restartTasksAction.failure({ alert }));
}
};
export const pauseConnector =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.pauseConnectorAction.request());
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action: ConnectorAction.PAUSE,
});
dispatch(actions.pauseConnectorAction.success({ connectorName }));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Pause`,
response,
};
dispatch(actions.pauseConnectorAction.failure({ alert }));
}
};
export const resumeConnector =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName
): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.resumeConnectorAction.request());
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action: ConnectorAction.RESUME,
});
dispatch(actions.resumeConnectorAction.success({ connectorName }));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Resume`,
response,
};
dispatch(actions.resumeConnectorAction.failure({ alert }));
}
};
export const restartConnectorTask =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
taskId: TaskId['task']
): PromiseThunkResult<void> =>
async (dispatch) => {
dispatch(actions.restartConnectorTaskAction.request());
try {
await kafkaConnectApiClient.restartConnectorTask({
clusterName,
connectName,
connectorName,
taskId: Number(taskId),
});
dispatch(actions.restartConnectorTaskAction.success());
dispatch(
fetchConnectorTasks(clusterName, connectName, connectorName, true)
);
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName, taskId].join('-'),
title: `Kafka Connect Connector Task Restart`,
response,
};
dispatch(actions.restartConnectorTaskAction.failure({ alert }));
}
};
export const fetchConnectorConfig =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
silent = false
): PromiseThunkResult<void> =>
async (dispatch) => {
if (!silent) dispatch(actions.fetchConnectorConfigAction.request());
try {
const config = await kafkaConnectApiClient.getConnectorConfig({
clusterName,
connectName,
connectorName,
});
dispatch(actions.fetchConnectorConfigAction.success({ config }));
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Config`,
response,
};
dispatch(actions.fetchConnectorConfigAction.failure({ alert }));
}
};
export const updateConnectorConfig =
(
clusterName: ClusterName,
connectName: ConnectName,
connectorName: ConnectorName,
connectorConfig: ConnectorConfig
): PromiseThunkResult<Connector | undefined> =>
async (dispatch) => {
dispatch(actions.updateConnectorConfigAction.request());
try {
const connector = await kafkaConnectApiClient.setConnectorConfig({
clusterName,
connectName,
connectorName,
requestBody: connectorConfig,
});
dispatch(actions.updateConnectorConfigAction.success({ connector }));
return connector;
} catch (error) {
const response = await getResponse(error);
const alert: FailurePayload = {
subject: [clusterName, connectName, connectorName].join('-'),
title: `Kafka Connect Connector Config Update`,
response,
};
dispatch(actions.updateConnectorConfigAction.failure({ alert }));
}
return undefined;
};
export const setConnectorSearch = (
connectorSearch: ConnectorSearch,
silent = false
): PromiseThunkResult<void> => {
return fetchConnectors(
connectorSearch.clusterName,
connectorSearch.search,
silent
);
};

1
kafka-ui-react-app/src/redux/actions/thunks/index.ts
View file

@ -1,2 +1 @@
export * from './topics';
export * from './connectors';

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

@ -1,19 +1,37 @@
import { ConnectorState, ConnectorTaskStatus } from 'generated-sources';
import {
fetchConnectorsAction,
fetchConnectorAction,
fetchConnectsAction,
fetchConnectorTasksAction,
fetchConnectorConfigAction,
createConnectorAction,
deleteConnectorAction,
pauseConnectorAction,
resumeConnectorAction,
updateConnectorConfigAction,
} from 'redux/actions';
import reducer, { initialState } from 'redux/reducers/connect/reducer';
ConnectorState,
ConnectorTaskStatus,
ConnectorAction,
} from 'generated-sources';
import reducer, {
initialState,
fetchConnects,
fetchConnectors,
fetchConnector,
createConnector,
deleteConnector,
setConnectorStatusState,
fetchConnectorTasks,
fetchConnectorConfig,
updateConnectorConfig,
restartConnector,
pauseConnector,
resumeConnector,
restartConnectorTask,
} from 'redux/reducers/connect/connectSlice';
import fetchMock from 'fetch-mock-jest';
import mockStoreCreator from 'redux/store/configureStore/mockStoreCreator';
import { getTypeAndPayload } from 'lib/testHelpers';
import { connects, connectors, connector, tasks } from './fixtures';
import {
connects,
connectors,
connector,
tasks,
connectorsServerPayload,
connectorServerPayload,
tasksServerPayload,
} from './fixtures';
const runningConnectorState = {
...initialState,
@ -57,28 +75,38 @@ const pausedConnectorState = {
},
};
describe('Clusters reducer', () => {
it('reacts on GET_CONNECTS__SUCCESS', () => {
describe('Connect slice', () => {
describe('Reducer', () => {
it('reacts on fetchConnects/fulfilled', () => {
expect(
reducer(initialState, fetchConnectsAction.success({ connects }))
reducer(initialState, {
type: fetchConnects.fulfilled,
payload: { connects },
})
).toEqual({
...initialState,
connects,
});
});
it('reacts on GET_CONNECTORS__SUCCESS', () => {
it('reacts on fetchConnectors/fulfilled', () => {
expect(
reducer(initialState, fetchConnectorsAction.success({ connectors }))
reducer(initialState, {
type: fetchConnectors.fulfilled,
payload: { connectors },
})
).toEqual({
...initialState,
connectors,
});
});
it('reacts on GET_CONNECTOR__SUCCESS', () => {
it('reacts on fetchConnector/fulfilled', () => {
expect(
reducer(initialState, fetchConnectorAction.success({ connector }))
reducer(initialState, {
type: fetchConnector.fulfilled,
payload: { connector },
})
).toEqual({
...initialState,
currentConnector: {
@ -88,9 +116,12 @@ describe('Clusters reducer', () => {
});
});
it('reacts on POST_CONNECTOR__SUCCESS', () => {
it('reacts on createConnector/fulfilled', () => {
expect(
reducer(initialState, createConnectorAction.success({ connector }))
reducer(initialState, {
type: createConnector.fulfilled,
payload: { connector },
})
).toEqual({
...initialState,
currentConnector: {
@ -100,14 +131,14 @@ describe('Clusters reducer', () => {
});
});
it('reacts on DELETE_CONNECTOR__SUCCESS', () => {
it('reacts on deleteConnector/fulfilled', () => {
expect(
reducer(
{ ...initialState, connectors },
{
...initialState,
connectors,
},
deleteConnectorAction.success({ connectorName: connectors[0].name })
type: deleteConnector.fulfilled,
payload: { connectorName: connectors[0].name },
}
)
).toEqual({
...initialState,
@ -115,69 +146,24 @@ describe('Clusters reducer', () => {
});
});
it('reacts on PAUSE_CONNECTOR__SUCCESS', () => {
it('reacts on setConnectorStatusState/fulfilled', () => {
expect(
reducer(
runningConnectorState,
pauseConnectorAction.success({ connectorName: connector.name })
)
reducer(runningConnectorState, {
type: setConnectorStatusState,
payload: {
taskState: ConnectorTaskStatus.PAUSED,
connectorState: ConnectorState.PAUSED,
},
})
).toEqual(pausedConnectorState);
});
it('reacts on PAUSE_CONNECTOR__SUCCESS when current connector is null', () => {
it('reacts on fetchConnectorTasks/fulfilled', () => {
expect(
reducer(
{
...initialState,
currentConnector: {
...initialState.currentConnector,
connector: null,
},
},
pauseConnectorAction.success({ connectorName: connector.name })
)
).toEqual({
...initialState,
currentConnector: {
...initialState.currentConnector,
connector: null,
},
});
});
it('reacts on RESUME_CONNECTOR__SUCCESS', () => {
expect(
reducer(
pausedConnectorState,
resumeConnectorAction.success({ connectorName: connector.name })
)
).toEqual(runningConnectorState);
});
it('reacts on RESUME_CONNECTOR__SUCCESS when current connector is null', () => {
expect(
reducer(
{
...initialState,
currentConnector: {
...initialState.currentConnector,
connector: null,
},
},
resumeConnectorAction.success({ connectorName: connector.name })
)
).toEqual({
...initialState,
currentConnector: {
...initialState.currentConnector,
connector: null,
},
});
});
it('reacts on GET_CONNECTOR_TASKS__SUCCESS', () => {
expect(
reducer(initialState, fetchConnectorTasksAction.success({ tasks }))
reducer(initialState, {
type: fetchConnectorTasks.fulfilled,
payload: { tasks },
})
).toEqual({
...initialState,
currentConnector: {
@ -187,12 +173,12 @@ describe('Clusters reducer', () => {
});
});
it('reacts on GET_CONNECTOR_CONFIG__SUCCESS', () => {
it('reacts on fetchConnectorConfig/fulfilled', () => {
expect(
reducer(
initialState,
fetchConnectorConfigAction.success({ config: connector.config })
)
reducer(initialState, {
type: fetchConnectorConfig.fulfilled,
payload: { config: connector.config },
})
).toEqual({
...initialState,
currentConnector: {
@ -202,7 +188,7 @@ describe('Clusters reducer', () => {
});
});
it('reacts on PATCH_CONNECTOR_CONFIG__SUCCESS', () => {
it('reacts on updateConnectorConfig/fulfilled', () => {
expect(
reducer(
{
@ -215,7 +201,10 @@ describe('Clusters reducer', () => {
},
},
},
updateConnectorConfigAction.success({ connector })
{
type: updateConnectorConfig.fulfilled,
payload: { connector },
}
)
).toEqual({
...initialState,
@ -227,3 +216,543 @@ describe('Clusters reducer', () => {
});
});
});
describe('Thunks', () => {
const store = mockStoreCreator;
const clusterName = 'local';
const connectName = 'first';
const connectorName = 'hdfs-source-connector';
const taskId = 10;
describe('Thunks', () => {
afterEach(() => {
fetchMock.restore();
store.clearActions();
});
describe('fetchConnects', () => {
it('creates fetchConnects/fulfilled when fetching connects', async () => {
fetchMock.getOnce(`/api/clusters/${clusterName}/connects`, connects);
await store.dispatch(fetchConnects(clusterName));
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnects.pending.type },
{
type: fetchConnects.fulfilled.type,
payload: { connects },
},
]);
});
it('creates fetchConnects/rejected', async () => {
fetchMock.getOnce(`/api/clusters/${clusterName}/connects`, 404);
await store.dispatch(fetchConnects(clusterName));
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnects.pending.type },
{
type: fetchConnects.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects`,
message: undefined,
},
},
]);
});
});
describe('fetchConnectors', () => {
it('creates fetchConnectors/fulfilled when fetching connectors', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connectors`,
connectorsServerPayload,
{ query: { search: '' } }
);
await store.dispatch(fetchConnectors({ clusterName }));
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnectors.pending.type },
{
type: fetchConnectors.fulfilled.type,
payload: { connectors },
},
]);
});
it('creates fetchConnectors/rejected', async () => {
fetchMock.getOnce(`/api/clusters/${clusterName}/connectors`, 404, {
query: { search: '' },
});
await store.dispatch(fetchConnectors({ clusterName }));
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnectors.pending.type },
{
type: fetchConnectors.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connectors?search=`,
message: undefined,
},
},
]);
});
});
describe('fetchConnector', () => {
it('creates fetchConnector/fulfilled when fetching connector', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
connectorServerPayload
);
await store.dispatch(
fetchConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnector.pending.type },
{
type: fetchConnector.fulfilled.type,
payload: { connector },
},
]);
});
it('creates fetchConnector/rejected', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
404
);
await store.dispatch(
fetchConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnector.pending.type },
{
type: fetchConnector.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
message: undefined,
},
},
]);
});
});
describe('createConnector', () => {
it('creates createConnector/fulfilled when fetching connects', async () => {
fetchMock.postOnce(
{
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors`,
body: {
name: connectorName,
config: connector.config,
},
},
connectorServerPayload
);
await store.dispatch(
createConnector({
clusterName,
connectName,
newConnector: {
name: connectorName,
config: connector.config,
},
})
);
expect(getTypeAndPayload(store)).toEqual([
{ type: createConnector.pending.type },
{
type: createConnector.fulfilled.type,
payload: { connector },
},
]);
});
it('creates createConnector/rejected', async () => {
fetchMock.postOnce(
{
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors`,
body: {
name: connectorName,
config: connector.config,
},
},
404
);
await store.dispatch(
createConnector({
clusterName,
connectName,
newConnector: {
name: connectorName,
config: connector.config,
},
})
);
expect(getTypeAndPayload(store)).toEqual([
{ type: createConnector.pending.type },
{
type: createConnector.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors`,
message: undefined,
},
},
]);
});
});
describe('deleteConnector', () => {
it('creates deleteConnector/fulfilled', async () => {
fetchMock.deleteOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
{}
);
fetchMock.getOnce(
`/api/clusters/${clusterName}/connectors?search=`,
connectorsServerPayload
);
await store.dispatch(
deleteConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: deleteConnector.pending.type },
{ type: fetchConnectors.pending.type },
{
type: deleteConnector.fulfilled.type,
payload: { connectorName },
},
]);
});
it('creates deleteConnector/rejected', async () => {
fetchMock.deleteOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
404
);
try {
await store.dispatch(
deleteConnector({ clusterName, connectName, connectorName })
);
} catch {
expect(getTypeAndPayload(store)).toEqual([
{ type: deleteConnector.pending.type },
{
type: deleteConnector.rejected.type,
payload: {
alert: {
subject: 'local-first-hdfs-source-connector',
title: 'Kafka Connect Connector Delete',
response: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}`,
},
},
},
},
]);
}
});
});
describe('fetchConnectorTasks', () => {
it('creates fetchConnectorTasks/fulfilled when fetching connects', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
tasksServerPayload
);
await store.dispatch(
fetchConnectorTasks({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnectorTasks.pending.type },
{
type: fetchConnectorTasks.fulfilled.type,
payload: { tasks },
},
]);
});
it('creates fetchConnectorTasks/rejected', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
404
);
await store.dispatch(
fetchConnectorTasks({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnectorTasks.pending.type },
{
type: fetchConnectorTasks.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
message: undefined,
},
},
]);
});
});
describe('restartConnector', () => {
it('creates restartConnector/fulfilled', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESTART}`,
{ message: 'success' }
);
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
tasksServerPayload
);
await store.dispatch(
restartConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: restartConnector.pending.type },
{ type: fetchConnectorTasks.pending.type },
{ type: restartConnector.fulfilled.type },
]);
});
it('creates restartConnector/rejected', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESTART}`,
404
);
await store.dispatch(
restartConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: restartConnector.pending.type },
{
type: restartConnector.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESTART}`,
message: undefined,
},
},
]);
});
});
describe('pauseConnector', () => {
it('creates pauseConnector/fulfilled when fetching connects', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.PAUSE}`,
{ message: 'success' }
);
await store.dispatch(
pauseConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: pauseConnector.pending.type },
{
type: setConnectorStatusState.type,
payload: {
connectorState: ConnectorState.PAUSED,
taskState: ConnectorTaskStatus.PAUSED,
},
},
{ type: pauseConnector.fulfilled.type },
]);
});
it('creates pauseConnector/rejected', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.PAUSE}`,
404
);
await store.dispatch(
pauseConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: pauseConnector.pending.type },
{
type: pauseConnector.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.PAUSE}`,
},
},
]);
});
});
describe('resumeConnector', () => {
it('creates resumeConnector/fulfilled when fetching connects', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESUME}`,
{ message: 'success' }
);
await store.dispatch(
resumeConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: resumeConnector.pending.type },
{
type: setConnectorStatusState.type,
payload: {
connectorState: ConnectorState.RUNNING,
taskState: ConnectorTaskStatus.RUNNING,
},
},
{ type: resumeConnector.fulfilled.type },
]);
});
it('creates resumeConnector/rejected', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESUME}`,
404
);
await store.dispatch(
resumeConnector({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: resumeConnector.pending.type },
{
type: resumeConnector.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/action/${ConnectorAction.RESUME}`,
},
},
]);
});
});
describe('restartConnectorTask', () => {
it('creates restartConnectorTask/fulfilled when fetching connects', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks/${taskId}/action/restart`,
{ message: 'success' }
);
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks`,
tasksServerPayload
);
await store.dispatch(
restartConnectorTask({
clusterName,
connectName,
connectorName,
taskId,
})
);
expect(getTypeAndPayload(store)).toEqual([
{ type: restartConnectorTask.pending.type },
{ type: fetchConnectorTasks.pending.type },
{ type: restartConnectorTask.fulfilled.type },
]);
});
it('creates restartConnectorTask/rejected', async () => {
fetchMock.postOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks/${taskId}/action/restart`,
404
);
await store.dispatch(
restartConnectorTask({
clusterName,
connectName,
connectorName,
taskId,
})
);
expect(getTypeAndPayload(store)).toEqual([
{ type: restartConnectorTask.pending.type },
{
type: restartConnectorTask.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/tasks/${taskId}/action/restart`,
},
},
]);
});
});
describe('fetchConnectorConfig', () => {
it('creates fetchConnectorConfig/fulfilled when fetching connects', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
connector.config
);
await store.dispatch(
fetchConnectorConfig({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnectorConfig.pending.type },
{
type: fetchConnectorConfig.fulfilled.type,
payload: { config: connector.config },
},
]);
});
it('creates fetchConnectorConfig/rejected', async () => {
fetchMock.getOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
404
);
await store.dispatch(
fetchConnectorConfig({ clusterName, connectName, connectorName })
);
expect(getTypeAndPayload(store)).toEqual([
{ type: fetchConnectorConfig.pending.type },
{
type: fetchConnectorConfig.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
message: undefined,
},
},
]);
});
});
describe('updateConnectorConfig', () => {
it('creates updateConnectorConfig/fulfilled when fetching connects', async () => {
fetchMock.putOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
connectorServerPayload
);
await store.dispatch(
updateConnectorConfig({
clusterName,
connectName,
connectorName,
connectorConfig: connector.config,
})
);
expect(getTypeAndPayload(store)).toEqual([
{ type: updateConnectorConfig.pending.type },
{
type: updateConnectorConfig.fulfilled.type,
payload: { connector },
},
]);
});
it('creates updateConnectorConfig/rejected', async () => {
fetchMock.putOnce(
`/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
404
);
await store.dispatch(
updateConnectorConfig({
clusterName,
connectName,
connectorName,
connectorConfig: connector.config,
})
);
expect(getTypeAndPayload(store)).toEqual([
{ type: updateConnectorConfig.pending.type },
{
type: updateConnectorConfig.rejected.type,
payload: {
status: 404,
statusText: 'Not Found',
url: `/api/clusters/${clusterName}/connects/${connectName}/connectors/${connectorName}/config`,
message: undefined,
},
},
]);
});
});
});
});
});

39
kafka-ui-react-app/src/redux/reducers/connect/__test__/selectors.spec.ts
View file

@ -1,10 +1,10 @@
import {
fetchConnectorAction,
fetchConnectorConfigAction,
fetchConnectorsAction,
fetchConnectorTasksAction,
fetchConnectsAction,
} from 'redux/actions';
fetchConnector,
fetchConnectorConfig,
fetchConnectors,
fetchConnectorTasks,
fetchConnects,
} from 'redux/reducers/connect/connectSlice';
import { store } from 'redux/store';
import * as selectors from 'redux/reducers/connect/selectors';
@ -50,17 +50,26 @@ describe('Connect selectors', () => {
describe('state', () => {
it('returns connects', () => {
store.dispatch(fetchConnectsAction.success({ connects }));
store.dispatch({
type: fetchConnects.fulfilled.type,
payload: { connects },
});
expect(selectors.getConnects(store.getState())).toEqual(connects);
});
it('returns connectors', () => {
store.dispatch(fetchConnectorsAction.success({ connectors }));
store.dispatch({
type: fetchConnectors.fulfilled.type,
payload: { connectors },
});
expect(selectors.getConnectors(store.getState())).toEqual(connectors);
});
it('returns connector', () => {
store.dispatch(fetchConnectorAction.success({ connector }));
store.dispatch({
type: fetchConnector.fulfilled.type,
payload: { connector },
});
expect(selectors.getConnector(store.getState())).toEqual(connector);
expect(selectors.getConnectorStatus(store.getState())).toEqual(
connector.status.state
@ -68,7 +77,10 @@ describe('Connect selectors', () => {
});
it('returns connector tasks', () => {
store.dispatch(fetchConnectorTasksAction.success({ tasks }));
store.dispatch({
type: fetchConnectorTasks.fulfilled.type,
payload: { tasks },
});
expect(selectors.getConnectorTasks(store.getState())).toEqual(tasks);
expect(selectors.getConnectorRunningTasksCount(store.getState())).toEqual(
2
@ -79,9 +91,10 @@ describe('Connect selectors', () => {
});
it('returns connector config', () => {
store.dispatch(
fetchConnectorConfigAction.success({ config: connector.config })
);
store.dispatch({
type: fetchConnectorConfig.fulfilled.type,
payload: { config: connector.config },
});
expect(selectors.getConnectorConfig(store.getState())).toEqual(
connector.config
);

468
kafka-ui-react-app/src/redux/reducers/connect/connectSlice.ts Normal file
View file

@ -0,0 +1,468 @@
import { createAsyncThunk, createSlice } from '@reduxjs/toolkit';
import {
Configuration,
Connect,
Connector,
ConnectorAction,
ConnectorState,
ConnectorTaskStatus,
FullConnectorInfo,
KafkaConnectApi,
NewConnector,
Task,
TaskId,
} from 'generated-sources';
import { BASE_PARAMS } from 'lib/constants';
import { getResponse } from 'lib/errorHandling';
import {
ClusterName,
ConnectName,
ConnectorConfig,
ConnectorName,
ConnectorSearch,
ConnectState,
} from 'redux/interfaces';
const apiClientConf = new Configuration(BASE_PARAMS);
export const kafkaConnectApiClient = new KafkaConnectApi(apiClientConf);
export const fetchConnects = createAsyncThunk<
{ connects: Connect[] },
ClusterName
>('connect/fetchConnects', async (clusterName, { rejectWithValue }) => {
try {
const connects = await kafkaConnectApiClient.getConnects({ clusterName });
return { connects };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
});
export const fetchConnectors = createAsyncThunk<
{ connectors: FullConnectorInfo[] },
{ clusterName: ClusterName; search?: string }
>(
'connect/fetchConnectors',
async ({ clusterName, search = '' }, { rejectWithValue }) => {
try {
const connectors = await kafkaConnectApiClient.getAllConnectors({
clusterName,
search,
});
return { connectors };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const fetchConnector = createAsyncThunk<
{ connector: Connector },
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}
>(
'connect/fetchConnector',
async ({ clusterName, connectName, connectorName }, { rejectWithValue }) => {
try {
const connector = await kafkaConnectApiClient.getConnector({
clusterName,
connectName,
connectorName,
});
return { connector };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const createConnector = createAsyncThunk<
{ connector: Connector },
{
clusterName: ClusterName;
connectName: ConnectName;
newConnector: NewConnector;
}
>(
'connect/createConnector',
async ({ clusterName, connectName, newConnector }, { rejectWithValue }) => {
try {
const connector = await kafkaConnectApiClient.createConnector({
clusterName,
connectName,
newConnector,
});
return { connector };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const deleteConnector = createAsyncThunk<
{ connectorName: string },
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}
>(
'connect/deleteConnector',
async (
{ clusterName, connectName, connectorName },
{ rejectWithValue, dispatch }
) => {
try {
await kafkaConnectApiClient.deleteConnector({
clusterName,
connectName,
connectorName,
});
dispatch(fetchConnectors({ clusterName, search: '' }));
return { connectorName };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const fetchConnectorTasks = createAsyncThunk<
{ tasks: Task[] },
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}
>(
'connect/fetchConnectorTasks',
async ({ clusterName, connectName, connectorName }, { rejectWithValue }) => {
try {
const tasks = await kafkaConnectApiClient.getConnectorTasks({
clusterName,
connectName,
connectorName,
});
return { tasks };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const restartConnector = createAsyncThunk<
undefined,
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}
>(
'connect/restartConnector',
async (
{ clusterName, connectName, connectorName },
{ rejectWithValue, dispatch }
) => {
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action: ConnectorAction.RESTART,
});
dispatch(
fetchConnectorTasks({
clusterName,
connectName,
connectorName,
})
);
return undefined;
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const restartTasks = createAsyncThunk<
undefined,
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
action: ConnectorAction;
}
>(
'connect/restartTasks',
async (
{ clusterName, connectName, connectorName, action },
{ rejectWithValue, dispatch }
) => {
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action,
});
dispatch(
fetchConnectorTasks({
clusterName,
connectName,
connectorName,
})
);
return undefined;
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const restartConnectorTask = createAsyncThunk<
undefined,
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
taskId: TaskId['task'];
}
>(
'connect/restartConnectorTask',
async (
{ clusterName, connectName, connectorName, taskId },
{ rejectWithValue, dispatch }
) => {
try {
await kafkaConnectApiClient.restartConnectorTask({
clusterName,
connectName,
connectorName,
taskId: Number(taskId),
});
dispatch(
fetchConnectorTasks({
clusterName,
connectName,
connectorName,
})
);
return undefined;
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const fetchConnectorConfig = createAsyncThunk<
{ config: { [key: string]: unknown } },
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}
>(
'connect/fetchConnectorConfig',
async ({ clusterName, connectName, connectorName }, { rejectWithValue }) => {
try {
const config = await kafkaConnectApiClient.getConnectorConfig({
clusterName,
connectName,
connectorName,
});
return { config };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const updateConnectorConfig = createAsyncThunk<
{ connector: Connector },
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
connectorConfig: ConnectorConfig;
}
>(
'connect/updateConnectorConfig',
async (
{ clusterName, connectName, connectorName, connectorConfig },
{ rejectWithValue }
) => {
try {
const connector = await kafkaConnectApiClient.setConnectorConfig({
clusterName,
connectName,
connectorName,
requestBody: connectorConfig,
});
return { connector };
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const initialState: ConnectState = {
connects: [],
connectors: [],
currentConnector: {
connector: null,
tasks: [],
config: null,
},
search: '',
};
const connectSlice = createSlice({
name: 'connect',
initialState,
reducers: {
setConnectorStatusState: (state, { payload }) => {
const { connector, tasks } = state.currentConnector;
if (connector) {
connector.status.state = payload.connectorState;
}
state.currentConnector.tasks = tasks.map((task) => ({
...task,
status: {
...task.status,
state: payload.taskState,
},
}));
},
},
extraReducers: (builder) => {
builder.addCase(fetchConnects.fulfilled, (state, { payload }) => {
state.connects = payload.connects;
});
builder.addCase(fetchConnectors.fulfilled, (state, { payload }) => {
state.connectors = payload.connectors;
});
builder.addCase(fetchConnector.fulfilled, (state, { payload }) => {
state.currentConnector.connector = payload.connector;
});
builder.addCase(createConnector.fulfilled, (state, { payload }) => {
state.currentConnector.connector = payload.connector;
});
builder.addCase(deleteConnector.fulfilled, (state, { payload }) => {
state.connectors = state.connectors.filter(
({ name }) => name !== payload.connectorName
);
});
builder.addCase(fetchConnectorTasks.fulfilled, (state, { payload }) => {
state.currentConnector.tasks = payload.tasks;
});
builder.addCase(fetchConnectorConfig.fulfilled, (state, { payload }) => {
state.currentConnector.config = payload.config;
});
builder.addCase(updateConnectorConfig.fulfilled, (state, { payload }) => {
state.currentConnector.connector = payload.connector;
state.currentConnector.config = payload.connector.config;
});
},
});
export const { setConnectorStatusState } = connectSlice.actions;
export const pauseCurrentConnector = () =>
setConnectorStatusState({
connectorState: ConnectorState.PAUSED,
taskState: ConnectorTaskStatus.PAUSED,
});
export const resumeCurrentConnector = () =>
setConnectorStatusState({
connectorState: ConnectorState.RUNNING,
taskState: ConnectorTaskStatus.RUNNING,
});
export const pauseConnector = createAsyncThunk<
undefined,
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}
>(
'connect/pauseConnector',
async (
{ clusterName, connectName, connectorName },
{ rejectWithValue, dispatch }
) => {
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action: ConnectorAction.PAUSE,
});
dispatch(pauseCurrentConnector());
return undefined;
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const resumeConnector = createAsyncThunk<
undefined,
{
clusterName: ClusterName;
connectName: ConnectName;
connectorName: ConnectorName;
}
>(
'connect/resumeConnector',
async (
{ clusterName, connectName, connectorName },
{ rejectWithValue, dispatch }
) => {
try {
await kafkaConnectApiClient.updateConnectorState({
clusterName,
connectName,
connectorName,
action: ConnectorAction.RESUME,
});
dispatch(resumeCurrentConnector());
return undefined;
} catch (err) {
return rejectWithValue(await getResponse(err as Response));
}
}
);
export const setConnectorSearch = (connectorSearch: ConnectorSearch) => {
return fetchConnectors({
clusterName: connectorSearch.clusterName,
search: connectorSearch.search,
});
};
export default connectSlice.reducer;

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

@ -1,123 +0,0 @@
import { getType } from 'typesafe-actions';
import * as actions from 'redux/actions';
import { ConnectState } from 'redux/interfaces/connect';
import { Action } from 'redux/interfaces';
import { ConnectorState, ConnectorTaskStatus } from 'generated-sources';
export const initialState: ConnectState = {
connects: [],
connectors: [],
currentConnector: {
connector: null,
tasks: [],
config: null,
},
search: '',
};
// eslint-disable-next-line @typescript-eslint/default-param-last
const reducer = (state = initialState, action: Action): ConnectState => {
switch (action.type) {
case getType(actions.fetchConnectsAction.success):
return {
...state,
connects: action.payload.connects,
};
case getType(actions.fetchConnectorsAction.success):
return {
...state,
connectors: action.payload.connectors,
};
case getType(actions.fetchConnectorAction.success):
case getType(actions.createConnectorAction.success):
return {
...state,
currentConnector: {
...state.currentConnector,
connector: action.payload.connector,
},
};
case getType(actions.deleteConnectorAction.success):
return {
...state,
connectors: state?.connectors.filter(
({ name }) => name !== action.payload.connectorName
),
};
case getType(actions.pauseConnectorAction.success):
return {
...state,
currentConnector: {
...state.currentConnector,
connector: state.currentConnector.connector
? {
...state.currentConnector.connector,
status: {
...state.currentConnector.connector?.status,
state: ConnectorState.PAUSED,
},
}
: null,
tasks: state.currentConnector.tasks.map((task) => ({
...task,
status: {
...task.status,
state: ConnectorTaskStatus.PAUSED,
},
})),
},
};
case getType(actions.resumeConnectorAction.success):
return {
...state,
currentConnector: {
...state.currentConnector,
connector: state.currentConnector.connector
? {
...state.currentConnector.connector,
status: {
...state.currentConnector.connector?.status,
state: ConnectorState.RUNNING,
},
}
: null,
tasks: state.currentConnector.tasks.map((task) => ({
...task,
status: {
...task.status,
state: ConnectorTaskStatus.RUNNING,
},
})),
},
};
case getType(actions.fetchConnectorTasksAction.success):
return {
...state,
currentConnector: {
...state.currentConnector,
tasks: action.payload.tasks,
},
};
case getType(actions.fetchConnectorConfigAction.success):
return {
...state,
currentConnector: {
...state.currentConnector,
config: action.payload.config,
},
};
case getType(actions.updateConnectorConfigAction.success):
return {
...state,
currentConnector: {
...state.currentConnector,
connector: action.payload.connector,
config: action.payload.connector.config,
},
};
default:
return state;
}
};
export default reducer;

74
kafka-ui-react-app/src/redux/reducers/connect/selectors.ts
View file

@ -1,14 +1,28 @@
import { createSelector } from '@reduxjs/toolkit';
import { ConnectState, RootState } from 'redux/interfaces';
import { createLeagcyFetchingSelector } from 'redux/reducers/loader/selectors';
import { createFetchingSelector } from 'redux/reducers/loader/selectors';
import { ConnectorTaskStatus } from 'generated-sources';
import {
deleteConnector,
fetchConnector,
fetchConnectorConfig,
fetchConnectors,
fetchConnectorTasks,
fetchConnects,
pauseConnector,
restartConnector,
resumeConnector,
} from './connectSlice';
const connectState = ({ connect }: RootState): ConnectState => connect;
const getConnectsFetchingStatus = createLeagcyFetchingSelector('GET_CONNECTS');
const getConnectsFetchingStatus = createFetchingSelector(
fetchConnects.typePrefix
);
export const getAreConnectsFetching = createSelector(
getConnectsFetchingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
export const getConnects = createSelector(
@ -16,11 +30,12 @@ export const getConnects = createSelector(
({ connects }) => connects
);
const getConnectorsFetchingStatus =
createLeagcyFetchingSelector('GET_CONNECTORS');
const getConnectorsFetchingStatus = createFetchingSelector(
fetchConnectors.typePrefix
);
export const getAreConnectorsFetching = createSelector(
getConnectorsFetchingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
export const getConnectors = createSelector(
@ -28,11 +43,12 @@ export const getConnectors = createSelector(
({ connectors }) => connectors
);
const getConnectorFetchingStatus =
createLeagcyFetchingSelector('GET_CONNECTOR');
const getConnectorFetchingStatus = createFetchingSelector(
fetchConnector.typePrefix
);
export const getIsConnectorFetching = createSelector(
getConnectorFetchingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
const getCurrentConnector = createSelector(
@ -50,32 +66,36 @@ export const getConnectorStatus = createSelector(
(connector) => connector?.status?.state
);
const getConnectorDeletingStatus =
createLeagcyFetchingSelector('DELETE_CONNECTOR');
const getConnectorDeletingStatus = createFetchingSelector(
deleteConnector.typePrefix
);
export const getIsConnectorDeleting = createSelector(
getConnectorDeletingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
const getConnectorRestartingStatus =
createLeagcyFetchingSelector('RESTART_CONNECTOR');
const getConnectorRestartingStatus = createFetchingSelector(
restartConnector.typePrefix
);
export const getIsConnectorRestarting = createSelector(
getConnectorRestartingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
const getConnectorPausingStatus =
createLeagcyFetchingSelector('PAUSE_CONNECTOR');
const getConnectorPausingStatus = createFetchingSelector(
pauseConnector.typePrefix
);
export const getIsConnectorPausing = createSelector(
getConnectorPausingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
const getConnectorResumingStatus =
createLeagcyFetchingSelector('RESUME_CONNECTOR');
const getConnectorResumingStatus = createFetchingSelector(
resumeConnector.typePrefix
);
export const getIsConnectorResuming = createSelector(
getConnectorResumingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
export const getIsConnectorActionRunning = createSelector(
@ -85,12 +105,12 @@ export const getIsConnectorActionRunning = createSelector(
(restarting, pausing, resuming) => restarting || pausing || resuming
);
const getConnectorTasksFetchingStatus = createLeagcyFetchingSelector(
'GET_CONNECTOR_TASKS'
const getConnectorTasksFetchingStatus = createFetchingSelector(
fetchConnectorTasks.typePrefix
);
export const getAreConnectorTasksFetching = createSelector(
getConnectorTasksFetchingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
export const getConnectorTasks = createSelector(
@ -112,12 +132,12 @@ export const getConnectorFailedTasksCount = createSelector(
.length
);
const getConnectorConfigFetchingStatus = createLeagcyFetchingSelector(
'GET_CONNECTOR_CONFIG'
const getConnectorConfigFetchingStatus = createFetchingSelector(
fetchConnectorConfig.typePrefix
);
export const getIsConnectorConfigFetching = createSelector(
getConnectorConfigFetchingStatus,
(status) => status === 'fetching'
(status) => status === 'pending'
);
export const getConnectorConfig = createSelector(

2
kafka-ui-react-app/src/redux/reducers/index.ts
View file

@ -4,11 +4,11 @@ import loader from 'redux/reducers/loader/loaderSlice';
import brokers from 'redux/reducers/brokers/brokersSlice';
import alerts from 'redux/reducers/alerts/alertsSlice';
import schemas from 'redux/reducers/schemas/schemasSlice';
import connect from 'redux/reducers/connect/connectSlice';
import topics from './topics/reducer';
import topicMessages from './topicMessages/reducer';
import consumerGroups from './consumerGroups/consumerGroupsSlice';
import connect from './connect/reducer';
import ksqlDb from './ksqlDb/ksqlDbSlice';
import legacyLoader from './loader/reducer';
import legacyAlerts from './alerts/reducer';