diff --git a/.github/playwright/impact-map.json b/.github/playwright/impact-map.json
index 920f402bbace..59279e569b6e 100644
--- a/.github/playwright/impact-map.json
+++ b/.github/playwright/impact-map.json
@@ -446,6 +446,34 @@
"playwright/e2e/Features/MutuallyExclusiveColumnTags.spec.ts",
"playwright/e2e/Features/ColumnBulkOperationsTagsGlossary.spec.ts"
]
+ },
+ {
+ "sources": [
+ "openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/**",
+ "openmetadata-ui/src/main/resources/ui/src/utils/TestConnectionModalUtils.tsx"
+ ],
+ "projects": ["Basic"],
+ "specs": ["playwright/e2e/Flow/TestConnectionModal.spec.ts"]
+ },
+ {
+ "sources": [
+ "openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/**",
+ "openmetadata-ui/src/main/resources/ui/src/utils/TestConnectionModalUtils.tsx"
+ ],
+ "projects": ["chromium"],
+ "specs": ["playwright/e2e/Flow/ConnectionConfigLayout.spec.ts"]
+ },
+ {
+ "sources": [
+ "openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/**",
+ "openmetadata-ui/src/main/resources/ui/src/utils/TestConnectionModalUtils.tsx"
+ ],
+ "projects": ["Ingestion"],
+ "specs": [
+ "playwright/e2e/Flow/ServiceForm.spec.ts",
+ "playwright/e2e/Flow/ApiServiceRest.spec.ts",
+ "playwright/e2e/Features/AutoPilot.spec.ts"
+ ]
}
]
}
diff --git a/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.test.tsx b/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.test.tsx
index d4049e37c77e..896666f8ff7e 100644
--- a/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.test.tsx
+++ b/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.test.tsx
@@ -1176,4 +1176,157 @@ describe('Test Connection Component', () => {
expect(onTestConnectionStatusChange).toHaveBeenLastCalledWith(true);
});
+
+ it('Should not let a superseded run overwrite the result of a newer one', async () => {
+ jest.useFakeTimers();
+
+ const onTestConnectionStatusChange = jest.fn();
+
+ // the first run dies while polling, which leaves its expiry timer armed
+ (getWorkflowById as jest.Mock).mockRejectedValueOnce(new Error('failed'));
+
+ await act(async () => {
+ render(
+
+ );
+ });
+
+ await act(async () => {
+ fireEvent.click(screen.getByTestId('test-connection-btn'));
+ });
+
+ await act(async () => {
+ jest.advanceTimersByTime(2000);
+ });
+
+ expect(
+ screen.getByText('message.connection-test-failed')
+ ).toBeInTheDocument();
+
+ await act(async () => {
+ fireEvent.click(screen.getByTestId('test-connection-btn'));
+ });
+
+ await act(async () => {
+ jest.advanceTimersByTime(2000);
+ });
+
+ expect(onTestConnectionStatusChange).toHaveBeenLastCalledWith(true);
+
+ // past the first run's expiry, whose callback must no longer touch the UI
+ await act(async () => {
+ jest.advanceTimersByTime(180000);
+ });
+
+ expect(onTestConnectionStatusChange).toHaveBeenLastCalledWith(true);
+ expect(
+ screen.queryByTestId('connection-timeout-message')
+ ).not.toBeInTheDocument();
+ });
+
+ it('Should ignore a superseded run whose poll resolves after a newer run started', async () => {
+ jest.useFakeTimers();
+
+ let resolveStalePoll: (value: unknown) => void = () => undefined;
+ const stalePoll = new Promise((resolve) => {
+ resolveStalePoll = resolve;
+ });
+
+ (getWorkflowById as jest.Mock)
+ .mockImplementationOnce(() => stalePoll)
+ .mockImplementation(() =>
+ Promise.resolve({
+ ...WORKFLOW_DETAILS,
+ status: 'Running',
+ response: { ...WORKFLOW_DETAILS.response, status: 'Running' },
+ })
+ );
+
+ await act(async () => {
+ render();
+ });
+
+ await act(async () => {
+ fireEvent.click(screen.getByTestId('test-connection-btn'));
+ });
+
+ // the first run's poll is now in flight and will not settle yet
+ await act(async () => {
+ jest.advanceTimersByTime(2000);
+ });
+
+ // its expiry re-enables the button while that request is still pending
+ await act(async () => {
+ jest.advanceTimersByTime(180000);
+ });
+
+ await act(async () => {
+ fireEvent.click(screen.getByTestId('test-connection-btn'));
+ });
+
+ await act(async () => {
+ jest.advanceTimersByTime(2000);
+ });
+
+ const callsBeforeStaleResolves = (getWorkflowById as jest.Mock).mock.calls
+ .length;
+
+ await act(async () => {
+ resolveStalePoll({
+ ...WORKFLOW_DETAILS,
+ status: 'Failed',
+ response: { ...WORKFLOW_DETAILS.response, status: 'Failed' },
+ });
+ });
+
+ expect(
+ screen.queryByText('message.connection-test-failed')
+ ).not.toBeInTheDocument();
+
+ // the newer run must still own its timers and keep polling
+ await act(async () => {
+ jest.advanceTimersByTime(2000);
+ });
+
+ expect((getWorkflowById as jest.Mock).mock.calls.length).toBeGreaterThan(
+ callsBeforeStaleResolves
+ );
+ });
+
+ it('Should stop polling the workflow after unmount', async () => {
+ jest.useFakeTimers();
+
+ (getWorkflowById as jest.Mock).mockImplementation(() =>
+ Promise.resolve({
+ ...WORKFLOW_DETAILS,
+ status: 'Running',
+ response: { ...WORKFLOW_DETAILS.response, status: 'Running' },
+ })
+ );
+
+ const { unmount } = render();
+
+ await act(async () => {
+ fireEvent.click(screen.getByTestId('test-connection-btn'));
+ });
+
+ await act(async () => {
+ jest.advanceTimersByTime(2000);
+ });
+
+ unmount();
+
+ const callsAtUnmount = (getWorkflowById as jest.Mock).mock.calls.length;
+
+ await act(async () => {
+ jest.advanceTimersByTime(180000);
+ });
+
+ expect((getWorkflowById as jest.Mock).mock.calls).toHaveLength(
+ callsAtUnmount
+ );
+ });
});
diff --git a/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.tsx b/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.tsx
index 7ba0aa22d9ff..cc5f3cfcd83c 100644
--- a/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.tsx
+++ b/openmetadata-ui/src/main/resources/ui/src/components/common/TestConnection/TestConnection.tsx
@@ -168,6 +168,17 @@ const TestConnection: FC = ({
*/
const currentWorkflowRef = useRef(currentWorkflow);
+ // Timers live in a ref so a new run - or an unmount - can cancel the previous
+ // run's callbacks before they write state over a newer result.
+ const timersRef = useRef<{ intervalId?: number; timeoutId?: number }>({});
+ const runIdRef = useRef(0);
+
+ const clearTimers = useCallback(() => {
+ clearInterval(timersRef.current.intervalId);
+ clearTimeout(timersRef.current.timeoutId);
+ timersRef.current = {};
+ }, []);
+
const { controller } = useAbortController();
const serviceType = useMemo(() => {
@@ -326,10 +337,7 @@ const TestConnection: FC = ({
const handleWorkflowPolling = async (
response: Workflow,
definitionSteps: TestConnectionStep[],
- intervalObject: {
- intervalId?: number;
- timeoutId?: number;
- }
+ runId: number
) => {
// return a promise that wraps the interval and handles errors inside it
return new Promise((resolve, reject) => {
@@ -337,14 +345,28 @@ const TestConnection: FC = ({
* fetch workflow repeatedly with 2s interval
* until status is either Failed or Successful
*/
- intervalObject.intervalId = toNumber(
+ timersRef.current.intervalId = toNumber(
setInterval(async () => {
+ if (runId !== runIdRef.current) {
+ resolve();
+
+ return;
+ }
+
setProgress(updateProgress);
try {
const workflowResponse = await getWorkflowData(
response.id,
controller.signal
);
+
+ // a newer run may have started while the request was in flight
+ if (runId !== runIdRef.current) {
+ resolve();
+
+ return;
+ }
+
const { response: testConnectionResponse } = workflowResponse;
const { status: testConnectionStatus, steps = [] } =
testConnectionResponse || {};
@@ -367,9 +389,7 @@ const TestConnection: FC = ({
definitionSteps
);
- // clear the current interval
- clearInterval(intervalObject.intervalId);
- clearTimeout(intervalObject.timeoutId);
+ clearTimers();
// set testing connection to false
setIsTestingConnection(false);
@@ -393,13 +413,10 @@ const TestConnection: FC = ({
setMessage(t(TEST_CONNECTION_TESTING_MESSAGE));
handleResetState();
- const updatedFormData = formatFormDataForSubmit(getData());
+ clearTimers();
+ const runId = ++runIdRef.current;
- // current interval id
- const intervalObject: {
- intervalId?: number;
- timeoutId?: number;
- } = {};
+ const updatedFormData = formatFormDataForSubmit(getData());
const { ingestionRunner, ...rest } = updatedFormData as ConfigObject & {
ingestionRunner?: string;
@@ -453,8 +470,12 @@ const TestConnection: FC = ({
// stop fetching the workflow after 2 minutes
const timeoutId = setTimeout(() => {
+ if (runId !== runIdRef.current) {
+ return;
+ }
+
// clear the current interval
- clearInterval(intervalObject.intervalId);
+ clearInterval(timersRef.current.intervalId);
// using reference to ensure call back should have latest value
const currentWorkflowStatus = currentWorkflowRef.current
@@ -482,13 +503,13 @@ const TestConnection: FC = ({
onTestConnectionStatusChange?.(false);
}, FETCHING_EXPIRY_TIME);
- intervalObject.timeoutId = Number(timeoutId);
+ timersRef.current.timeoutId = Number(timeoutId);
// Handle workflow polling and completion
- await handleWorkflowPolling(response, definitionSteps, intervalObject);
+ await handleWorkflowPolling(response, definitionSteps, runId);
} catch (error) {
setProgress(TEST_CONNECTION_PROGRESS_PERCENTAGE.HUNDRED);
- clearInterval(intervalObject.intervalId);
+ clearTimers();
setIsTestingConnection(false);
setMessage(t(TEST_CONNECTION_FAILURE_MESSAGE));
setTestStatus(StatusType.Failed);
@@ -678,6 +699,8 @@ const TestConnection: FC = ({
useEffect(() => {
return () => {
+ clearTimers();
+
/**
* if workflow is present then delete the workflow when component unmount
*/