Copilot commented on code in PR #17696: URL: https://github.com/apache/dolphinscheduler/pull/17696#discussion_r2562575036
########## dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/test/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTaskTest.java: ########## @@ -0,0 +1,352 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.plugin.task.procedure; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor; +import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider; +import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider; +import org.apache.dolphinscheduler.plugin.task.api.TaskConstants; +import org.apache.dolphinscheduler.plugin.task.api.TaskException; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy; +import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper; +import org.apache.dolphinscheduler.spi.datasource.ConnectionParam; +import org.apache.dolphinscheduler.spi.enums.DbType; + +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.sql.CallableStatement; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.Map; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class ProcedureTaskTest { + + @Mock + private TaskExecutionContext mockTaskContext; + + @Mock + private ResourceParametersHelper mockResourceHelper; + + @Mock + private ProcedureParameters procedureParams = new ProcedureParameters(); + + private ProcedureTask procedureTask; + + @BeforeEach + void setUp() { + + // Simulate valid task parameters JSON + String validParamsJson = "{" + + "\"type\":\"MYSQL\"," + + "\"datasource\":\"test_db\"," + + "\"method\":\"{call my_proc(?, ?)}\"," + + "\"localParams\":{}," + + "\"outProperty\":{}" + + "}"; + + when(mockTaskContext.getTaskParams()).thenReturn(validParamsJson); + when(mockTaskContext.getResourceParametersHelper()).thenReturn(mockResourceHelper); + + // Mock parameter validation to pass + try (MockedStatic<JSONUtils> jsonUtilsMock = mockStatic(JSONUtils.class)) { + jsonUtilsMock.when(() -> JSONUtils.parseObject(anyString(), eq(ProcedureParameters.class))) + .thenReturn(procedureParams); + doCallRealMethod().when(procedureParams).checkParameters(); // or stub as needed + when(procedureParams.checkParameters()).thenReturn(true); + when(procedureParams.generateExtendedContext(any())).thenReturn(mock(ProcedureTaskExecutionContext.class)); + + procedureTask = new ProcedureTask(mockTaskContext); + } + + } + + @Test + void constructor_InvalidParameters_ThrowsTaskException() { + // Arrange: invalid JSON or checkParameters returns false + String invalidParamsJson = "{}"; + when(mockTaskContext.getTaskParams()).thenReturn(invalidParamsJson); + + try (MockedStatic<JSONUtils> jsonUtilsMock = mockStatic(JSONUtils.class)) { + ProcedureParameters badParams = mock(ProcedureParameters.class); + jsonUtilsMock.when(() -> JSONUtils.parseObject(anyString(), eq(ProcedureParameters.class))) + .thenReturn(badParams); + when(badParams.checkParameters()).thenReturn(false); + + // Act & Assert + TaskException exception = assertThrows(TaskException.class, + () -> new ProcedureTask(mockTaskContext)); + assertTrue(exception.getMessage().contains("not valid")); + } + } + + @Test + void handle_SuccessfulExecution_SetsSuccessExitCode() throws Exception { + when(procedureParams.getType()).thenReturn("MYSQL"); + when(procedureParams.getMethod()).thenReturn("{call my_proc(?, ?)}"); + + // Arrange mocks + Connection mockConn = mock(Connection.class); + CallableStatement mockStmt = mock(CallableStatement.class); + + try ( + MockedStatic<DataSourceClientProvider> clientMock = mockStatic(DataSourceClientProvider.class); + MockedStatic<DataSourceProcessorProvider> processorMock = + mockStatic(DataSourceProcessorProvider.class)) { + + DataSourceProcessor mockProcessor = mock(DataSourceProcessor.class); + ConnectionParam mockConnParam = mock(ConnectionParam.class); + + processorMock.when(() -> DataSourceProcessorProvider.getDataSourceProcessor(DbType.MYSQL)) + .thenReturn(mockProcessor); + when(mockProcessor.createConnectionParams((String) null)).thenReturn(mockConnParam); + clientMock.when(() -> DataSourceClientProvider.getAdHocConnection(eq(DbType.MYSQL), any())) + .thenReturn(mockConn); + + when(mockConn.prepareCall(anyString())).thenReturn(mockStmt); + when(mockStmt.executeUpdate()).thenReturn(1); // success + + // Mock parameter maps to avoid NPE + when(mockTaskContext.getPrepareParamsMap()).thenReturn(new HashMap<>()); + + // Act + procedureTask.handle(null); + + // Assert + assertEquals(TaskConstants.EXIT_CODE_SUCCESS, getField(procedureTask, "exitStatusCode")); + verify(mockStmt).executeUpdate(); + } + } + + @Test + void handle_SqlException_SetsFailureExitCodeAndThrows() throws SQLException { + when(procedureParams.getType()).thenReturn("MYSQL"); + when(procedureParams.getMethod()).thenReturn("{call my_proc(?, ?)}"); + + // Arrange + Connection mockConn = mock(Connection.class); + CallableStatement mockStmt = mock(CallableStatement.class); + + try ( + MockedStatic<DataSourceClientProvider> clientMock = mockStatic(DataSourceClientProvider.class); + MockedStatic<DataSourceProcessorProvider> processorMock = + mockStatic(DataSourceProcessorProvider.class)) { + + DataSourceProcessor mockProcessor = mock(DataSourceProcessor.class); + ConnectionParam mockConnParam = mock(ConnectionParam.class); + + processorMock.when(() -> DataSourceProcessorProvider.getDataSourceProcessor(DbType.MYSQL)) + .thenReturn(mockProcessor); + when(mockProcessor.createConnectionParams((String) null)).thenReturn(mockConnParam); + clientMock.when(() -> DataSourceClientProvider.getAdHocConnection(eq(DbType.MYSQL), any())) + .thenReturn(mockConn); + + when(mockConn.prepareCall(anyString())).thenReturn(mockStmt); + when(mockStmt.executeUpdate()).thenThrow(new SQLException("DB error")); + + when(mockTaskContext.getPrepareParamsMap()).thenReturn(new HashMap<>()); + + // Act & Assert + TaskException exception = assertThrows(TaskException.class, + () -> procedureTask.handle(null)); + assertTrue(exception.getMessage().contains("failed")); + assertEquals(TaskConstants.EXIT_CODE_FAILURE, getField(procedureTask, "exitStatusCode")); + } + } + + @Test + void cancel_ActiveStatement_CancelsAndSetsKillCode() throws Exception { + // Arrange: simulate active statement + CallableStatement mockStmt = mock(CallableStatement.class); + setField(procedureTask, "sessionStatement", mockStmt); + + // Act + procedureTask.cancel(); + + // Assert + verify(mockStmt).cancel(); + assertEquals(TaskConstants.EXIT_CODE_KILL, getField(procedureTask, "exitStatusCode")); + } + + @Test + void cancel_NoActiveStatement_LogsWarning() { + // sessionStatement is null by default after construction + // We just call cancel and ensure no exception + assertDoesNotThrow(() -> procedureTask.cancel()); + // (In practice, you'd verify logger.warn was called — requires Logback/TestLogger setup) + } + + @Test + void cancel_whenStatementCancelThrowsSQLException_shouldLogWarningAndThrowTaskException() throws SQLException { + // Arrange: simulate active statement + CallableStatement mockStmt = mock(CallableStatement.class); + setField(procedureTask, "sessionStatement", mockStmt); + + // Arrange + SQLException sqlEx = new SQLException("Driver does not support cancel"); + doThrow(sqlEx).when(mockStmt).cancel(); + + // Act & Assert + TaskException taskEx = assertThrows(TaskException.class, () -> { + procedureTask.cancel(); + }); + + // Verify the cause is the original SQLException + assertEquals(sqlEx, taskEx.getCause()); + + // Verify logging behavior indirectly (optional: use MockLogger for strict verification) + // Here we assume logging occurs as per implementation + + // Verify that exit status was NOT set (since stmt.cancel() threw before setExitStatusCode) + // You may need a getter or reflection to check internal state + Integer exitCode = getPrivateField(procedureTask, "exitStatusCode"); + assertEquals(exitCode, -1); Review Comment: The test expects exitStatusCode to be -1 when cancel() throws an exception, but this contradicts the bug found in the implementation. According to line 144 of ProcedureTask.java, if stmt.cancel() throws an exception, setExitStatusCode() is never called, so the exit code remains at its previous value. The test assumption that it should be -1 is incorrect unless that's the default value. Consider verifying what the actual initial/default exit status code is, or update this test to reflect the actual behavior after fixing the implementation bug. ```suggestion Integer exitCodeBefore = getPrivateField(procedureTask, "exitStatusCode"); Integer exitCodeAfter = getPrivateField(procedureTask, "exitStatusCode"); assertEquals(exitCodeAfter, exitCodeBefore, "exitStatusCode should remain unchanged when cancel() throws"); ``` ########## dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java: ########## @@ -105,30 +108,49 @@ public void handle(TaskCallBack taskCallBack) throws TaskException { } String proceduerSql = formatSql(sqlParamsMap, paramsMap); // call method - try (CallableStatement stmt = connection.prepareCall(proceduerSql)) { + try (CallableStatement tmpStatement = connection.prepareCall(proceduerSql)) { + sessionStatement = tmpStatement; // set timeout - setTimeout(stmt); + setTimeout(tmpStatement); // outParameterMap - Map<Integer, Property> outParameterMap = getOutParameterMap(stmt, sqlParamsMap, paramsMap); + Map<Integer, Property> outParameterMap = getOutParameterMap(tmpStatement, sqlParamsMap, paramsMap); - stmt.executeUpdate(); + tmpStatement.executeUpdate(); // print the output parameters to the log - printOutParameter(stmt, outParameterMap); + printOutParameter(tmpStatement, outParameterMap); setExitStatusCode(EXIT_CODE_SUCCESS); } } catch (Exception e) { + if (exitStatusCode == TaskConstants.EXIT_CODE_KILL) { + log.info("procedure task has been killed"); + return; + } setExitStatusCode(EXIT_CODE_FAILURE); log.error("procedure task error", e); throw new TaskException("Execute procedure task failed", e); } } @Override - public void cancel() throws TaskException { - + public void cancel() { + Statement stmt = this.sessionStatement; + if (stmt != null) { + try { + log.debug("Try to cancel this procedure task"); + stmt.cancel(); + setExitStatusCode(TaskConstants.EXIT_CODE_KILL); + log.debug("this procedure task was canceled"); + } catch (SQLException ex) { + log.warn("Failed to cancel stored procedure (driver/DB may not support it)", ex); + throw new TaskException("Cancel http task failed", ex); Review Comment: The error message refers to "http task" but this is a procedure task. The message should be "Cancel procedure task failed" instead of "Cancel http task failed". ```suggestion throw new TaskException("Cancel procedure task failed", ex); ``` ########## dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java: ########## @@ -105,30 +108,49 @@ public void handle(TaskCallBack taskCallBack) throws TaskException { } String proceduerSql = formatSql(sqlParamsMap, paramsMap); // call method - try (CallableStatement stmt = connection.prepareCall(proceduerSql)) { + try (CallableStatement tmpStatement = connection.prepareCall(proceduerSql)) { + sessionStatement = tmpStatement; // set timeout - setTimeout(stmt); + setTimeout(tmpStatement); // outParameterMap - Map<Integer, Property> outParameterMap = getOutParameterMap(stmt, sqlParamsMap, paramsMap); + Map<Integer, Property> outParameterMap = getOutParameterMap(tmpStatement, sqlParamsMap, paramsMap); - stmt.executeUpdate(); + tmpStatement.executeUpdate(); // print the output parameters to the log - printOutParameter(stmt, outParameterMap); + printOutParameter(tmpStatement, outParameterMap); setExitStatusCode(EXIT_CODE_SUCCESS); } } catch (Exception e) { + if (exitStatusCode == TaskConstants.EXIT_CODE_KILL) { + log.info("procedure task has been killed"); + return; + } setExitStatusCode(EXIT_CODE_FAILURE); log.error("procedure task error", e); throw new TaskException("Execute procedure task failed", e); } } @Override - public void cancel() throws TaskException { - + public void cancel() { Review Comment: The cancel() method signature was changed to remove the `throws TaskException` declaration, which is inconsistent with the parent class AbstractTask that declares `public abstract void cancel() throws TaskException;`. Even though the implementation throws TaskException on line 148, the signature should maintain the throws declaration for consistency with the parent class and other task implementations. ```suggestion public void cancel() throws TaskException { ``` ########## dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java: ########## @@ -105,30 +108,49 @@ public void handle(TaskCallBack taskCallBack) throws TaskException { } String proceduerSql = formatSql(sqlParamsMap, paramsMap); // call method - try (CallableStatement stmt = connection.prepareCall(proceduerSql)) { + try (CallableStatement tmpStatement = connection.prepareCall(proceduerSql)) { + sessionStatement = tmpStatement; // set timeout - setTimeout(stmt); + setTimeout(tmpStatement); // outParameterMap - Map<Integer, Property> outParameterMap = getOutParameterMap(stmt, sqlParamsMap, paramsMap); + Map<Integer, Property> outParameterMap = getOutParameterMap(tmpStatement, sqlParamsMap, paramsMap); - stmt.executeUpdate(); + tmpStatement.executeUpdate(); // print the output parameters to the log - printOutParameter(stmt, outParameterMap); + printOutParameter(tmpStatement, outParameterMap); setExitStatusCode(EXIT_CODE_SUCCESS); } } catch (Exception e) { + if (exitStatusCode == TaskConstants.EXIT_CODE_KILL) { + log.info("procedure task has been killed"); + return; + } setExitStatusCode(EXIT_CODE_FAILURE); log.error("procedure task error", e); throw new TaskException("Execute procedure task failed", e); } } @Override - public void cancel() throws TaskException { - + public void cancel() { + Statement stmt = this.sessionStatement; + if (stmt != null) { + try { + log.debug("Try to cancel this procedure task"); + stmt.cancel(); + setExitStatusCode(TaskConstants.EXIT_CODE_KILL); Review Comment: The exit status code should be set to EXIT_CODE_KILL before attempting to cancel the statement. If stmt.cancel() throws an exception, the exit status will not be set, which could lead to incorrect task status reporting. Consider moving line 144 before line 143. ```suggestion setExitStatusCode(TaskConstants.EXIT_CODE_KILL); stmt.cancel(); ``` ########## dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/test/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTaskTest.java: ########## @@ -0,0 +1,352 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.plugin.task.procedure; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor; +import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider; +import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider; +import org.apache.dolphinscheduler.plugin.task.api.TaskConstants; +import org.apache.dolphinscheduler.plugin.task.api.TaskException; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy; +import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper; +import org.apache.dolphinscheduler.spi.datasource.ConnectionParam; +import org.apache.dolphinscheduler.spi.enums.DbType; + +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.sql.CallableStatement; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.Map; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class ProcedureTaskTest { + + @Mock + private TaskExecutionContext mockTaskContext; + + @Mock + private ResourceParametersHelper mockResourceHelper; + + @Mock + private ProcedureParameters procedureParams = new ProcedureParameters(); Review Comment: The @Mock annotation should not be used with initialization. On line 73, `procedureParams` is annotated with @Mock but also initialized with `= new ProcedureParameters()`. This is contradictory - either use @Mock alone or remove the annotation and just instantiate. The initialization will be overwritten by Mockito anyway. ```suggestion private ProcedureParameters procedureParams; ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
