@@ -428,6 +437,14 @@ export default defineComponent({
onViewLog={handleViewLog}
onExecuteTask={handleExecuteTask}
onRemoveTaskInstanceCache={handleRemoveTaskInstanceCache}
+ v-model:dependenciesData={dependenciesData}
+ />
+
{!!props.definition && (
{
if (task.code === currTask.value?.code) {
setNodeName(task.code + '', taskDef.name)
let fillColor = '#ffffff'
- if (task.flag === 'YES') {
+ if (taskDef.flag === 'NO') {
fillColor = 'var(--custom-disable-bg)'
}
+
setNodeFillColor(task.code + '', fillColor)
setNodeEdge(String(task.code), data.preTasks)
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/definition/components/table-action.tsx b/dolphinscheduler-ui/src/views/projects/workflow/definition/components/table-action.tsx
index 0cc08bc44c..1453e7ed4f 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/definition/components/table-action.tsx
+++ b/dolphinscheduler-ui/src/views/projects/workflow/definition/components/table-action.tsx
@@ -33,6 +33,7 @@ import {
} from '@vicons/antd'
import { useI18n } from 'vue-i18n'
import { IDefinitionData } from '../types'
+
const props = {
row: {
type: Object as PropType
@@ -95,6 +96,7 @@ export default defineComponent({
const handleReleaseScheduler = () => {
ctx.emit('releaseScheduler')
}
+
return {
handleEditWorkflow,
handleStartWorkflow,
@@ -114,6 +116,7 @@ export default defineComponent({
const releaseState = this.row?.releaseState
const scheduleReleaseState = this.row?.scheduleReleaseState
const schedule = this.row?.schedule
+
return (
@@ -166,10 +169,7 @@ export default defineComponent({
trigger: () => (
{{
- default: () =>
- releaseState === 'ONLINE'
- ? t('project.workflow.confirm_to_offline')
- : t('project.workflow.confirm_to_online'),
+ default: () => releaseState === 'OFFLINE' ? t('project.workflow.confirm_to_online'):t('project.workflow.confirm_to_offline'),
trigger: () => (
{{
default: () =>
- scheduleReleaseState === 'ONLINE'
- ? t('project.workflow.time_to_offline')
- : t('project.workflow.time_to_online'),
+ scheduleReleaseState === 'OFFLINE' ? t('project.workflow.time_to_online'):t('project.workflow.time_to_offline'),
trigger: () => (
{
- queryAllWorkerGroups().then((res: any) => {
- variables.workerGroups = res.map((item: string) => ({
- label: item,
- value: item
- }))
+ queryWorkerGroupsByProjectCode(variables.projectCode).then((res: any) => {
+ variables.workerGroups = res.data.map((item: any) =>({label: item.workerGroup, value: item.workerGroup }))
})
}
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/definition/index.tsx b/dolphinscheduler-ui/src/views/projects/workflow/definition/index.tsx
index 3eff91673f..90b55adf17 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/definition/index.tsx
+++ b/dolphinscheduler-ui/src/views/projects/workflow/definition/index.tsx
@@ -24,7 +24,7 @@ import {
NSpace,
NTooltip,
NPopconfirm,
- NModal
+ NModal,
} from 'naive-ui'
import {
defineComponent,
@@ -45,6 +45,7 @@ import VersionModal from './components/version-modal'
import CopyModal from './components/copy-modal'
import type { Router } from 'vue-router'
import Search from '@/components/input-search'
+import DependenciesModal from '@/views/projects/components/dependencies/dependencies-modal'
export default defineComponent({
name: 'WorkflowDefinitionList',
@@ -318,6 +319,14 @@ export default defineComponent({
maskClosable={false}
onPositiveClick={this.confirmToSetWorkflowTiming}
/>
+
)
}
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/index.tsx b/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/index.tsx
index bd348fb8d7..9450138e0f 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/index.tsx
+++ b/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/index.tsx
@@ -24,6 +24,7 @@ import { useTable } from './use-table'
import Card from '@/components/card'
import TimingModal from '../components/timing-modal'
import type { Router } from 'vue-router'
+import DependenciesModal from "@/views/projects/components/dependencies/dependencies-modal";
export default defineComponent({
name: 'WorkflowDefinitionTiming',
@@ -115,6 +116,13 @@ export default defineComponent({
v-model:show={this.showRef}
onUpdateList={this.handleUpdateList}
/>
+
)
}
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/use-table.ts b/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/use-table.ts
index a339e20e46..4c86637d04 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/use-table.ts
+++ b/dolphinscheduler-ui/src/views/projects/workflow/definition/timing/use-table.ts
@@ -39,15 +39,18 @@ import {
import { format } from 'date-fns-tz'
import { ISearchParam } from './types'
import type { Router } from 'vue-router'
+import { useDependencies } from "@/views/projects/components/dependencies/use-dependencies"
export function useTable() {
const { t } = useI18n()
const router: Router = useRouter()
+ const {getDependentTaskLinks} = useDependencies()
+
const variables = reactive({
columns: [],
tableWidth: DefaultTableWidth,
- row: {},
+ row: {} as any,
tableData: [],
projectCode: ref(Number(router.currentRoute.value.params.projectCode)),
page: ref(1),
@@ -58,7 +61,8 @@ export function useTable() {
loadingRef: ref(false),
processDefinitionCode: router.currentRoute.value.params.definitionCode
? ref(Number(router.currentRoute.value.params.definitionCode))
- : ref()
+ : ref(),
+ dependenciesData: ref({showRef: false, taskLinks: ref([]), required: ref(false), tip: ref(''), action:() => {}}),
})
const renderTime = (time: string, timeZone: string) => {
@@ -329,7 +333,7 @@ export function useTable() {
NPopconfirm,
{
onPositiveClick: () => {
- handleDelete(row.id)
+ handleDelete(row)
}
},
{
@@ -344,7 +348,8 @@ export function useTable() {
{
circle: true,
type: 'error',
- size: 'small'
+ size: 'small',
+ disabled: row.releaseState === 'ONLINE'
},
{
icon: () => h(DeleteOutlined)
@@ -387,12 +392,43 @@ export function useTable() {
}
const handleReleaseState = (row: any) => {
- let handle = online
if (row.releaseState === 'ONLINE') {
- handle = offline
+ variables.row = row
+ getDependentTaskLinks(variables.projectCode, row.processDefinitionCode).then((res: any) =>{
+ if (res && res.length > 0) {
+ variables.dependenciesData.showRef = true
+ variables.dependenciesData.taskLinks = res
+ variables.dependenciesData.tip = t('project.workflow.warning_delete_scheduler_dependent_tasks_desc')
+ variables.dependenciesData.required = false
+ variables.dependenciesData.action = confirmToOfflineSchedule
+ } else {
+ offline(variables.projectCode, row.id).then(() => {
+ window.$message.success(t('project.workflow.success'))
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal,
+ projectCode: variables.projectCode,
+ processDefinitionCode: variables.processDefinitionCode
+ })
+ })
+ }})
+ } else {
+ online(variables.projectCode, row.id).then(() => {
+ window.$message.success(t('project.workflow.success'))
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal,
+ projectCode: variables.projectCode,
+ processDefinitionCode: variables.processDefinitionCode
+ })
+ })
}
+ }
- handle(variables.projectCode, row.id).then(() => {
+ const confirmToOfflineSchedule = () => {
+ offline(variables.projectCode, variables.row.id).then(() => {
window.$message.success(t('project.workflow.success'))
getTableData({
pageSize: variables.pageSize,
@@ -402,14 +438,11 @@ export function useTable() {
processDefinitionCode: variables.processDefinitionCode
})
})
+ variables.dependenciesData.showRef = false
}
- const handleDelete = (id: number) => {
- /* after deleting data from the current page, you need to jump forward when the page is empty. */
- if (variables.tableData.length === 1 && variables.page > 1) {
- variables.page -= 1
- }
- deleteScheduleById(id, variables.projectCode).then(() => {
+ const confirmToDeleteSchedule = () => {
+ deleteScheduleById(variables.row.id, variables.projectCode).then(() => {
window.$message.success(t('project.workflow.success'))
getTableData({
pageSize: variables.pageSize,
@@ -419,6 +452,35 @@ export function useTable() {
processDefinitionCode: variables.processDefinitionCode
})
})
+ variables.dependenciesData.showRef = false
+ }
+
+ const handleDelete = (row: any) => {
+ /* after deleting data from the current page, you need to jump forward when the page is empty. */
+ if (variables.tableData.length === 1 && variables.page > 1) {
+ variables.page -= 1
+ }
+ variables.row = row
+ getDependentTaskLinks(variables.projectCode, row.processDefinitionCode).then((res: any) =>{
+ if (res && res.length > 0) {
+ variables.dependenciesData.showRef = true
+ variables.dependenciesData.taskLinks = res
+ variables.dependenciesData.tip = t('project.workflow.warning_delete_scheduler_dependent_tasks_desc')
+ variables.dependenciesData.required = false
+ variables.dependenciesData.action = confirmToDeleteSchedule
+ } else {
+ deleteScheduleById(row.id, variables.projectCode).then(() => {
+ window.$message.success(t('project.workflow.success'))
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal,
+ projectCode: variables.projectCode,
+ processDefinitionCode: variables.processDefinitionCode
+ })
+ })
+ }
+ })
}
return {
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/definition/use-table.ts b/dolphinscheduler-ui/src/views/projects/workflow/definition/use-table.ts
index 914c10a15a..04228d532f 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/definition/use-table.ts
+++ b/dolphinscheduler-ui/src/views/projects/workflow/definition/use-table.ts
@@ -43,11 +43,14 @@ import {
import type { IDefinitionParam } from './types'
import type { Router } from 'vue-router'
import type { TableColumns, RowKey } from 'naive-ui/es/data-table/src/interface'
+import {useDependencies} from '../../components/dependencies/use-dependencies'
export function useTable() {
const { t } = useI18n()
const router: Router = useRouter()
const { copy } = useTextCopy()
+ const { getDependentTaskLinks } = useDependencies()
+
const variables = reactive({
columns: [],
tableWidth: DefaultTableWidth,
@@ -67,7 +70,8 @@ export function useTable() {
versionShowRef: ref(false),
copyShowRef: ref(false),
loadingRef: ref(false),
- setTimingDialogShowRef: ref(false)
+ setTimingDialogShowRef: ref(false),
+ dependenciesData: ref({showRef: false, taskLinks: ref([]), required: ref(false), tip: ref(''), action:() => {}}),
})
const createColumns = (variables: any) => {
@@ -304,17 +308,6 @@ export function useTable() {
variables.row = row
}
- const deleteWorkflow = (row: any) => {
- deleteByCode(variables.projectCode, row.code).then(() => {
- window.$message.success(t('project.workflow.success'))
- getTableData({
- pageSize: variables.pageSize,
- pageNo: variables.page,
- searchVal: variables.searchVal
- })
- })
- }
-
const batchDeleteWorkflow = () => {
const data = {
codes: _.join(variables.checkedRowKeys, ',')
@@ -354,43 +347,142 @@ export function useTable() {
const batchCopyWorkflow = () => {}
- const releaseWorkflow = (row: any) => {
+ const confirmToOfflineWorkflow = () => {
+ const row: any = variables.row
const data = {
name: row.name,
releaseState: (row.releaseState === 'ONLINE' ? 'OFFLINE' : 'ONLINE') as
- | 'OFFLINE'
- | 'ONLINE'
+ | 'OFFLINE'
+ | 'ONLINE'
}
-
- release(data, variables.projectCode, row.code).then(() => {
- if (data.releaseState === 'ONLINE') {
- variables.setTimingDialogShowRef = true
- variables.row = row
- } else {
+ if (data.releaseState === 'OFFLINE') {
+ release(data, variables.projectCode, row.code).then(() => {
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal
+ })
window.$message.success(t('project.workflow.success'))
- }
+ })
+ }
+ variables.dependenciesData.showRef = false
+ }
+
+ const confirmToOfflineScheduler = () => {
+ const row: any = variables.row
+ offline(variables.projectCode, row.schedule.id).then(() => {
+ window.$message.success(t('project.workflow.success'))
getTableData({
pageSize: variables.pageSize,
pageNo: variables.page,
searchVal: variables.searchVal
})
})
+ variables.dependenciesData.showRef = false
}
- const releaseScheduler = (row: any) => {
- if (row.schedule) {
- let handle = online
- if (row.schedule.releaseState === 'ONLINE') {
- handle = offline
- }
- handle(variables.projectCode, row.schedule.id).then(() => {
- window.$message.success(t('project.workflow.success'))
+ const releaseWorkflow = (row: any) => {
+ const data = {
+ name: row.name,
+ releaseState: (row.releaseState === 'ONLINE' ? 'OFFLINE' : 'ONLINE') as
+ | 'OFFLINE'
+ | 'ONLINE'
+ }
+ variables.row = row
+ if (data.releaseState === 'ONLINE') {
+ release(data, variables.projectCode, row.code).then(() => {
+ variables.setTimingDialogShowRef = true
+ if (row?.schedule) {
+ variables.row = row.schedule
+ variables.timingType = 'update'
+ variables.timingState = row.scheduleReleaseState
+ }
getTableData({
pageSize: variables.pageSize,
pageNo: variables.page,
searchVal: variables.searchVal
})
})
+ } else {
+ getDependentTaskLinks(variables.projectCode, row.code).then((res: any) => {
+ if (res && res.length > 0) {
+ variables.dependenciesData = {
+ showRef: true,
+ taskLinks: res,
+ tip: t('project.workflow.warning_dependent_tasks_desc'),
+ required: false,
+ action: confirmToOfflineWorkflow
+ }
+ } else {
+ release(data, variables.projectCode, row.code).then(() => {
+ window.$message.success(t('project.workflow.success'))
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal
+ })
+ })
+ }
+ })
+ }
+ }
+
+ const deleteWorkflow = (row: any) => {
+ getDependentTaskLinks(variables.projectCode, row.code).then((res: any) => {
+ if (res && res.length > 0) {
+ variables.dependenciesData = {
+ showRef: true,
+ taskLinks: res,
+ tip: t('project.workflow.delete_validate_dependent_tasks_desc'),
+ required: true,
+ action: () => {}
+ }
+ } else {
+ deleteByCode(variables.projectCode, row.code).then(() => {
+ window.$message.success(t('project.workflow.success'))
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal
+ })
+ })
+ }
+ })
+ }
+
+ const releaseScheduler = (row: any) => {
+ variables.row = row
+ if (row.schedule) {
+ if (row.schedule.releaseState === 'ONLINE') {
+ getDependentTaskLinks(variables.projectCode, row.code).then((res: any) => {
+ if (res && res.length > 0) {
+ variables.dependenciesData = {
+ showRef: true,
+ taskLinks: res,
+ tip: t('project.workflow.warning_offline_scheduler_dependent_tasks_desc'),
+ required: false,
+ action: confirmToOfflineScheduler
+ }
+ } else {
+ offline(variables.projectCode, row.schedule.id).then(() => {
+ window.$message.success(t('project.workflow.success'))
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal
+ })
+ })
+ }})
+ } else {
+ online(variables.projectCode, row.schedule.id).then(() => {
+ window.$message.success(t('project.workflow.success'))
+ getTableData({
+ pageSize: variables.pageSize,
+ pageNo: variables.page,
+ searchVal: variables.searchVal
+ })
+ })
+ }
}
}
@@ -474,6 +566,6 @@ export function useTable() {
getTableData,
batchDeleteWorkflow,
batchExportWorkflow,
- batchCopyWorkflow
+ batchCopyWorkflow,
}
}
diff --git a/dolphinscheduler-ui/src/views/projects/workflow/timing/index.tsx b/dolphinscheduler-ui/src/views/projects/workflow/timing/index.tsx
index 1228cfd835..37a33447d5 100644
--- a/dolphinscheduler-ui/src/views/projects/workflow/timing/index.tsx
+++ b/dolphinscheduler-ui/src/views/projects/workflow/timing/index.tsx
@@ -15,14 +15,15 @@
* limitations under the License.
*/
-import { NDataTable, NPagination, NSpace } from 'naive-ui'
-import { defineComponent, onMounted, toRefs, watch } from 'vue'
+import {NDataTable, NPagination, NSpace} from 'naive-ui'
+import {defineComponent, onMounted, toRefs, watch} from 'vue'
import { useI18n } from 'vue-i18n'
import { useTable } from '../definition/timing/use-table'
import Card from '@/components/card'
import TimingModal from '../definition/components/timing-modal'
import TimingCondition from '@/views/projects/workflow/timing/components/timing-condition'
import { ITimingSearch } from '@/views/projects/workflow/timing/types'
+import DependenciesModal from "@/views/projects/components/dependencies/dependencies-modal";
export default defineComponent({
name: 'WorkflowTimingList',
@@ -110,6 +111,13 @@ export default defineComponent({
v-model:show={this.showRef}
onUpdateList={this.handleUpdateList}
/>
+
)
}
diff --git a/dolphinscheduler-ui/src/views/resource/task-group/option/use-table.ts b/dolphinscheduler-ui/src/views/resource/task-group/option/use-table.ts
index fc286d7d0c..838a1d39cb 100644
--- a/dolphinscheduler-ui/src/views/resource/task-group/option/use-table.ts
+++ b/dolphinscheduler-ui/src/views/resource/task-group/option/use-table.ts
@@ -147,6 +147,7 @@ export function useTable(
parseTime(item.updateTime),
'yyyy-MM-dd HH:mm:ss'
)
+ item.status = (item.status == 'YES') ? 1 : 0
return {
...item
}
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorFactoryBuilder.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorFactoryBuilder.java
index a9c2948482..599746818d 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorFactoryBuilder.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorFactoryBuilder.java
@@ -48,6 +48,21 @@ public class WorkerTaskExecutorFactoryBuilder {
@Autowired
private WorkerRegistryClient workerRegistryClient;
+ public WorkerTaskExecutorFactoryBuilder(
+ WorkerConfig workerConfig,
+ WorkerMessageSender workerMessageSender,
+ TaskPluginManager taskPluginManager,
+ WorkerTaskExecutorThreadPool workerManager,
+ StorageOperate storageOperate,
+ WorkerRegistryClient workerRegistryClient) {
+ this.workerConfig = workerConfig;
+ this.workerMessageSender = workerMessageSender;
+ this.taskPluginManager = taskPluginManager;
+ this.workerManager = workerManager;
+ this.storageOperate = storageOperate;
+ this.workerRegistryClient = workerRegistryClient;
+ }
+
public WorkerTaskExecutorFactory extends WorkerTaskExecutor> createWorkerTaskExecutorFactory(TaskExecutionContext taskExecutionContext) {
return new DefaultWorkerTaskExecutorFactory(taskExecutionContext,
workerConfig,
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionEventAckListenFunctionManager.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionEventAckListenFunctionManager.java
index b4423f4880..3214be8c89 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionEventAckListenFunctionManager.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionEventAckListenFunctionManager.java
@@ -35,6 +35,15 @@ public class TaskInstanceExecutionEventAckListenFunctionManager {
@Autowired
private TaskInstanceExecutionInfoEventAckListenFunction taskInstanceExecutionInfoEventAckListenFunction;
+ public TaskInstanceExecutionEventAckListenFunctionManager(
+ TaskInstanceExecutionRunningEventAckListenFunction taskInstanceExecutionRunningEventAckListenFunction,
+ TaskInstanceExecutionFinishEventAckListenFunction taskInstanceExecutionFinishEventAckListenFunction,
+ TaskInstanceExecutionInfoEventAckListenFunction taskInstanceExecutionInfoEventAckListenFunction) {
+ this.taskInstanceExecutionRunningEventAckListenFunction = taskInstanceExecutionRunningEventAckListenFunction;
+ this.taskInstanceExecutionFinishEventAckListenFunction = taskInstanceExecutionFinishEventAckListenFunction;
+ this.taskInstanceExecutionInfoEventAckListenFunction = taskInstanceExecutionInfoEventAckListenFunction;
+ }
+
public TaskInstanceExecutionRunningEventAckListenFunction getTaskInstanceExecutionRunningEventAckListenFunction() {
return taskInstanceExecutionRunningEventAckListenFunction;
}
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionFinishEventAckListenFunction.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionFinishEventAckListenFunction.java
index ad7892bc7a..a358623519 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionFinishEventAckListenFunction.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionFinishEventAckListenFunction.java
@@ -36,6 +36,10 @@ public class TaskInstanceExecutionFinishEventAckListenFunction
@Autowired
private MessageRetryRunner messageRetryRunner;
+ public TaskInstanceExecutionFinishEventAckListenFunction(MessageRetryRunner messageRetryRunner) {
+ this.messageRetryRunner = messageRetryRunner;
+ }
+
@Override
public void handleTaskInstanceExecutionEventAck(TaskInstanceExecutionFinishEventAck taskInstanceExecutionFinishEventAck) {
try {
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionInfoEventAckListenFunction.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionInfoEventAckListenFunction.java
index 971343103a..b3dcc9bf8a 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionInfoEventAckListenFunction.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionInfoEventAckListenFunction.java
@@ -37,6 +37,10 @@ public class TaskInstanceExecutionInfoEventAckListenFunction
@Resource
private MessageRetryRunner messageRetryRunner;
+ public TaskInstanceExecutionInfoEventAckListenFunction(MessageRetryRunner messageRetryRunner) {
+ this.messageRetryRunner = messageRetryRunner;
+ }
+
@Override
public void handleTaskInstanceExecutionEventAck(TaskInstanceExecutionInfoEventAck taskInstanceExecutionInfoEventAck) {
try {
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionRunningEventAckListenFunction.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionRunningEventAckListenFunction.java
index 9d6de78e02..e17d72ad99 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionRunningEventAckListenFunction.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionRunningEventAckListenFunction.java
@@ -36,6 +36,9 @@ public class TaskInstanceExecutionRunningEventAckListenFunction
@Autowired
private MessageRetryRunner messageRetryRunner;
+ public TaskInstanceExecutionRunningEventAckListenFunction(MessageRetryRunner messageRetryRunner) {
+ this.messageRetryRunner = messageRetryRunner;
+ }
@Override
public void handleTaskInstanceExecutionEventAck(TaskInstanceExecutionRunningEventAck taskInstanceExecutionRunningEventAck) {
try {
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceDispatchOperationFunction.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceDispatchOperationFunction.java
index e6d259412f..fc128a9a34 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceDispatchOperationFunction.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceDispatchOperationFunction.java
@@ -48,6 +48,15 @@ public class TaskInstanceDispatchOperationFunction
@Autowired
private WorkerTaskExecutorThreadPool workerTaskExecutorThreadPool;
+ public TaskInstanceDispatchOperationFunction(
+ WorkerConfig workerConfig,
+ WorkerTaskExecutorFactoryBuilder workerTaskExecutorFactoryBuilder,
+ WorkerTaskExecutorThreadPool workerTaskExecutorThreadPool) {
+ this.workerConfig = workerConfig;
+ this.workerTaskExecutorFactoryBuilder = workerTaskExecutorFactoryBuilder;
+ this.workerTaskExecutorThreadPool = workerTaskExecutorThreadPool;
+ }
+
@Override
public TaskInstanceDispatchResponse operate(TaskInstanceDispatchRequest taskInstanceDispatchRequest) {
log.info("Receive TaskInstanceDispatchRequest: {}", taskInstanceDispatchRequest);
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceKillOperationFunction.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceKillOperationFunction.java
index 69e3994a90..d55765d23f 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceKillOperationFunction.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceKillOperationFunction.java
@@ -50,6 +50,13 @@ public class TaskInstanceKillOperationFunction
@Autowired
private MessageRetryRunner messageRetryRunner;
+ public TaskInstanceKillOperationFunction(
+ WorkerTaskExecutorThreadPool workerManager,
+ MessageRetryRunner messageRetryRunner) {
+ this.workerManager = workerManager;
+ this.messageRetryRunner = messageRetryRunner;
+ }
+
@Override
public TaskInstanceKillResponse operate(TaskInstanceKillRequest taskInstanceKillRequest) {
log.info("Receive TaskInstanceKillRequest: {}", taskInstanceKillRequest);
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionManager.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionManager.java
index 99ae193b47..8014b88fd1 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionManager.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionManager.java
@@ -35,6 +35,17 @@ public class TaskInstanceOperationFunctionManager {
@Autowired
private TaskInstancePauseOperationFunction taskInstancePauseOperationFunction;
+ public TaskInstanceOperationFunctionManager(
+ TaskInstanceKillOperationFunction taskInstanceKillOperationFunction,
+ UpdateWorkflowHostOperationFunction updateWorkflowHostOperationFunction,
+ TaskInstanceDispatchOperationFunction taskInstanceDispatchOperationFunction,
+ TaskInstancePauseOperationFunction taskInstancePauseOperationFunction) {
+ this.taskInstanceKillOperationFunction = taskInstanceKillOperationFunction;
+ this.updateWorkflowHostOperationFunction = updateWorkflowHostOperationFunction;
+ this.taskInstanceDispatchOperationFunction = taskInstanceDispatchOperationFunction;
+ this.taskInstancePauseOperationFunction = taskInstancePauseOperationFunction;
+ }
+
public TaskInstanceKillOperationFunction getTaskInstanceKillOperationFunction() {
return taskInstanceKillOperationFunction;
}
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/UpdateWorkflowHostOperationFunction.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/UpdateWorkflowHostOperationFunction.java
index 7485b9230f..c0ab345450 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/UpdateWorkflowHostOperationFunction.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/operator/UpdateWorkflowHostOperationFunction.java
@@ -39,6 +39,10 @@ public class UpdateWorkflowHostOperationFunction
@Autowired
private MessageRetryRunner messageRetryRunner;
+ public UpdateWorkflowHostOperationFunction(MessageRetryRunner messageRetryRunner) {
+ this.messageRetryRunner = messageRetryRunner;
+ }
+
@Override
public UpdateWorkflowHostResponse operate(UpdateWorkflowHostRequest updateWorkflowHostRequest) {
try {
diff --git a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionEventAckListenFunctionTest.java b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionEventAckListenFunctionTest.java
new file mode 100644
index 0000000000..5044fba11e
--- /dev/null
+++ b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/listener/TaskInstanceExecutionEventAckListenFunctionTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.server.worker.runner.listener;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.times;
+
+import org.apache.dolphinscheduler.extract.master.transportor.ITaskInstanceExecutionEvent;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceExecutionFinishEventAck;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceExecutionInfoEventAck;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceExecutionRunningEventAck;
+import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TaskInstanceExecutionEventAckListenFunctionTest {
+
+ private static final Logger log = LoggerFactory.getLogger(TaskInstanceExecutionEventAckListenFunctionTest.class);
+ private MessageRetryRunner messageRetryRunner = Mockito.mock(MessageRetryRunner.class);
+
+ @Test
+ public void testTaskInstanceExecutionEventAckListenFunctionManager() {
+ TaskInstanceExecutionFinishEventAckListenFunction taskInstanceExecutionFinishEventAckListenFunction =
+ new TaskInstanceExecutionFinishEventAckListenFunction(messageRetryRunner);
+ TaskInstanceExecutionInfoEventAckListenFunction taskInstanceExecutionInfoEventAckListenFunction =
+ new TaskInstanceExecutionInfoEventAckListenFunction(messageRetryRunner);
+ TaskInstanceExecutionRunningEventAckListenFunction taskInstanceExecutionRunningEventAckListenFunction =
+ new TaskInstanceExecutionRunningEventAckListenFunction(messageRetryRunner);
+ TaskInstanceExecutionEventAckListenFunctionManager taskInstanceExecutionEventAckListenFunctionManager =
+ new TaskInstanceExecutionEventAckListenFunctionManager(
+ taskInstanceExecutionRunningEventAckListenFunction,
+ taskInstanceExecutionFinishEventAckListenFunction,
+ taskInstanceExecutionInfoEventAckListenFunction);
+ Assertions.assertEquals(taskInstanceExecutionRunningEventAckListenFunction,
+ taskInstanceExecutionEventAckListenFunctionManager
+ .getTaskInstanceExecutionRunningEventAckListenFunction());
+ Assertions.assertEquals(taskInstanceExecutionInfoEventAckListenFunction,
+ taskInstanceExecutionEventAckListenFunctionManager
+ .getTaskInstanceExecutionInfoEventAckListenFunction());
+ Assertions.assertEquals(taskInstanceExecutionFinishEventAckListenFunction,
+ taskInstanceExecutionEventAckListenFunctionManager
+ .getTaskInstanceExecutionFinishEventAckListenFunction());
+ }
+
+ @Test
+ public void testTaskInstanceExecutionEventAckListenFunctionDryRun() {
+ int taskInstanceId1 = 111;
+ int taskInstanceId2 = 222;
+ int taskInstanceId3 = 333;
+ TaskInstanceExecutionFinishEventAckListenFunction taskInstanceExecutionFinishEventAckListenFunction =
+ new TaskInstanceExecutionFinishEventAckListenFunction(messageRetryRunner);
+ taskInstanceExecutionFinishEventAckListenFunction.handleTaskInstanceExecutionEventAck(
+ TaskInstanceExecutionFinishEventAck.success(taskInstanceId1));
+
+ ArgumentCaptor acInt = ArgumentCaptor.forClass(int.class);
+ ArgumentCaptor acEventType =
+ ArgumentCaptor.forClass(ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType.class);
+
+ Mockito.verify(messageRetryRunner, times(1)).removeRetryMessage(
+ (int) acInt.capture(),
+ (ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType) acEventType.capture());
+
+ assertEquals(taskInstanceId1, acInt.getValue());
+
+ TaskInstanceExecutionInfoEventAckListenFunction taskInstanceExecutionInfoEventAckListenFunction =
+ new TaskInstanceExecutionInfoEventAckListenFunction(messageRetryRunner);
+ taskInstanceExecutionInfoEventAckListenFunction.handleTaskInstanceExecutionEventAck(
+ TaskInstanceExecutionInfoEventAck.success(taskInstanceId2));
+
+ Mockito.verify(messageRetryRunner, times(2)).removeRetryMessage(
+ (int) acInt.capture(),
+ (ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType) acEventType.capture());
+ assertEquals(taskInstanceId2, acInt.getValue());
+
+ TaskInstanceExecutionRunningEventAckListenFunction taskInstanceExecutionRunningEventAckListenFunction =
+ new TaskInstanceExecutionRunningEventAckListenFunction(messageRetryRunner);
+ taskInstanceExecutionRunningEventAckListenFunction.handleTaskInstanceExecutionEventAck(
+ TaskInstanceExecutionRunningEventAck.success(taskInstanceId3));
+ Mockito.verify(messageRetryRunner, times(3)).removeRetryMessage(
+ (int) acInt.capture(),
+ (ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType) acEventType.capture());
+ assertEquals(taskInstanceId3, acInt.getValue());
+ }
+}
diff --git a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionTest.java b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionTest.java
new file mode 100644
index 0000000000..592340214f
--- /dev/null
+++ b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.server.worker.runner.operator;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceDispatchRequest;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceDispatchResponse;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceKillRequest;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstanceKillResponse;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstancePauseRequest;
+import org.apache.dolphinscheduler.extract.worker.transportor.TaskInstancePauseResponse;
+import org.apache.dolphinscheduler.extract.worker.transportor.UpdateWorkflowHostRequest;
+import org.apache.dolphinscheduler.extract.worker.transportor.UpdateWorkflowHostResponse;
+import org.apache.dolphinscheduler.plugin.storage.api.StorageOperate;
+import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
+import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
+import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
+import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner;
+import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
+import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
+import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecutor;
+import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecutorFactoryBuilder;
+import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecutorHolder;
+import org.apache.dolphinscheduler.server.worker.runner.WorkerTaskExecutorThreadPool;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TaskInstanceOperationFunctionTest {
+
+ private static final Logger log = LoggerFactory.getLogger(TaskInstanceOperationFunctionTest.class);
+ private MessageRetryRunner messageRetryRunner = Mockito.mock(MessageRetryRunner.class);
+
+ private WorkerConfig workerConfig = Mockito.mock(WorkerConfig.class);
+
+ private TaskExecutionContext taskExecutionContext = Mockito.mock(TaskExecutionContext.class);
+
+ private WorkerTaskExecutorThreadPool workerTaskExecutorThreadPool =
+ Mockito.mock(WorkerTaskExecutorThreadPool.class);
+
+ private WorkerTaskExecutor workerTaskExecutor = Mockito.mock(WorkerTaskExecutor.class);
+
+ private AbstractTask task = Mockito.mock(AbstractTask.class);
+
+ private WorkerMessageSender workerMessageSender = Mockito.mock(WorkerMessageSender.class);
+
+ private TaskPluginManager taskPluginManager = Mockito.mock(TaskPluginManager.class);
+
+ private WorkerTaskExecutorThreadPool workerManager = Mockito.mock(WorkerTaskExecutorThreadPool.class);
+
+ private StorageOperate storageOperate = Mockito.mock(StorageOperate.class);
+
+ private WorkerRegistryClient workerRegistryClient = Mockito.mock(WorkerRegistryClient.class);
+
+ @Test
+ public void testTaskInstanceOperationFunctionManager() {
+ TaskInstanceKillOperationFunction taskInstanceKillOperationFunction = new TaskInstanceKillOperationFunction(
+ workerTaskExecutorThreadPool,
+ messageRetryRunner);
+
+ TaskInstancePauseOperationFunction taskInstancePauseOperationFunction =
+ new TaskInstancePauseOperationFunction();
+
+ UpdateWorkflowHostOperationFunction updateWorkflowHostOperationFunction =
+ new UpdateWorkflowHostOperationFunction(
+ messageRetryRunner);
+
+ WorkerTaskExecutorFactoryBuilder workerTaskExecutorFactoryBuilder = new WorkerTaskExecutorFactoryBuilder(
+ workerConfig,
+ workerMessageSender,
+ taskPluginManager,
+ workerManager,
+ storageOperate,
+ workerRegistryClient);
+
+ TaskInstanceDispatchOperationFunction taskInstanceDispatchOperationFunction =
+ new TaskInstanceDispatchOperationFunction(
+ workerConfig,
+ workerTaskExecutorFactoryBuilder,
+ workerTaskExecutorThreadPool);
+
+ TaskInstanceOperationFunctionManager taskInstanceOperationFunctionManager =
+ new TaskInstanceOperationFunctionManager(
+ taskInstanceKillOperationFunction,
+ updateWorkflowHostOperationFunction,
+ taskInstanceDispatchOperationFunction,
+ taskInstancePauseOperationFunction);
+
+ Assertions.assertEquals(taskInstanceKillOperationFunction,
+ taskInstanceOperationFunctionManager.getTaskInstanceKillOperationFunction());
+ Assertions.assertEquals(taskInstancePauseOperationFunction,
+ taskInstanceOperationFunctionManager.getTaskInstancePauseOperationFunction());
+ Assertions.assertEquals(updateWorkflowHostOperationFunction,
+ taskInstanceOperationFunctionManager.getUpdateWorkflowHostOperationFunction());
+ Assertions.assertEquals(taskInstanceDispatchOperationFunction,
+ taskInstanceOperationFunctionManager.getTaskInstanceDispatchOperationFunction());
+ }
+
+ @Test
+ public void testUpdateWorkflowHostOperationFunction() {
+ UpdateWorkflowHostOperationFunction updateWorkflowHostOperationFunction =
+ new UpdateWorkflowHostOperationFunction(
+ messageRetryRunner);
+
+ try (MockedStatic logUtilsMockedStatic = Mockito.mockStatic(LogUtils.class)) {
+ logUtilsMockedStatic
+ .when(() -> LogUtils
+ .setTaskInstanceIdMDC(any(Integer.class)))
+ .then(invocationOnMock -> null);
+ UpdateWorkflowHostRequest request = new UpdateWorkflowHostRequest();
+ request.setTaskInstanceId(1);
+ request.setWorkflowHost("host");
+ UpdateWorkflowHostResponse taskInstanceDispatchResponse = updateWorkflowHostOperationFunction.operate(
+ request);
+ Assertions.assertEquals(taskInstanceDispatchResponse.isSuccess(), false);
+ }
+
+ try (MockedStatic logUtilsMockedStatic = Mockito.mockStatic(LogUtils.class)) {
+ logUtilsMockedStatic
+ .when(() -> LogUtils
+ .setTaskInstanceIdMDC(any(Integer.class)))
+ .then(invocationOnMock -> null);
+
+ try (
+ MockedStatic workerTaskExecutorHolderMockedStatic =
+ Mockito.mockStatic(WorkerTaskExecutorHolder.class)) {
+ given(workerTaskExecutor.getTaskExecutionContext()).willReturn(taskExecutionContext);
+ workerTaskExecutorHolderMockedStatic
+ .when(() -> WorkerTaskExecutorHolder.get(any(Integer.class)))
+ .thenReturn(workerTaskExecutor);
+ int taskInstanceId = 111;
+ UpdateWorkflowHostRequest request = new UpdateWorkflowHostRequest();
+ request.setTaskInstanceId(taskInstanceId);
+ request.setWorkflowHost("host");
+
+ UpdateWorkflowHostResponse taskInstanceDispatchResponse = updateWorkflowHostOperationFunction.operate(
+ request);
+ Assertions.assertEquals(taskInstanceDispatchResponse.isSuccess(), true);
+ }
+ }
+ }
+
+ @Test
+ public void testTaskInstancePauseOperationFunction() {
+ TaskInstancePauseOperationFunction taskInstancePauseOperationFunction =
+ new TaskInstancePauseOperationFunction();
+
+ try (MockedStatic logUtilsMockedStatic = Mockito.mockStatic(LogUtils.class)) {
+ logUtilsMockedStatic
+ .when(() -> LogUtils
+ .setTaskInstanceIdMDC(any(Integer.class)))
+ .then(invocationOnMock -> null);
+ TaskInstancePauseRequest request = new TaskInstancePauseRequest();
+ request.setTaskInstanceId(1);
+ TaskInstancePauseResponse taskInstanceDispatchResponse = taskInstancePauseOperationFunction.operate(
+ request);
+ Assertions.assertEquals(taskInstanceDispatchResponse.isSuccess(), true);
+ }
+ }
+
+ @Test
+ public void testTaskInstanceDispatchOperationFunction() {
+ WorkerTaskExecutorFactoryBuilder workerTaskExecutorFactoryBuilder = new WorkerTaskExecutorFactoryBuilder(
+ workerConfig,
+ workerMessageSender,
+ taskPluginManager,
+ workerManager,
+ storageOperate,
+ workerRegistryClient);
+
+ TaskInstanceDispatchOperationFunction taskInstanceDispatchOperationFunction =
+ new TaskInstanceDispatchOperationFunction(
+ workerConfig,
+ workerTaskExecutorFactoryBuilder,
+ workerTaskExecutorThreadPool);
+
+ try (MockedStatic logUtilsMockedStatic = Mockito.mockStatic(LogUtils.class)) {
+ logUtilsMockedStatic
+ .when(() -> LogUtils
+ .getTaskInstanceLogFullPath(any(TaskExecutionContext.class)))
+ .thenReturn("test");
+ TaskInstanceDispatchResponse taskInstanceDispatchResponse = taskInstanceDispatchOperationFunction.operate(
+ new TaskInstanceDispatchRequest(taskExecutionContext));
+ Assertions.assertEquals(taskInstanceDispatchResponse.isDispatchSuccess(), false);
+ logUtilsMockedStatic.verify(times(1), () -> LogUtils.removeWorkflowAndTaskInstanceIdMDC());
+
+ given(workerTaskExecutorThreadPool.submitWorkerTaskExecutor(any())).willReturn(true);
+ taskInstanceDispatchResponse = taskInstanceDispatchOperationFunction.operate(
+ new TaskInstanceDispatchRequest(taskExecutionContext));
+ Assertions.assertEquals(taskInstanceDispatchResponse.isDispatchSuccess(), true);
+ logUtilsMockedStatic.verify(times(2), () -> LogUtils.removeWorkflowAndTaskInstanceIdMDC());
+ }
+ }
+
+ @Test
+ public void testTaskInstanceKillOperationFunction() {
+ TaskInstanceKillOperationFunction taskInstanceKillOperationFunction = new TaskInstanceKillOperationFunction(
+ workerManager,
+ messageRetryRunner);
+
+ try (MockedStatic logUtilsMockedStatic = Mockito.mockStatic(LogUtils.class)) {
+ int taskInstanceId = 111;
+ logUtilsMockedStatic
+ .when(() -> LogUtils
+ .setTaskInstanceLogFullPathMDC(any(String.class)))
+ .then(invocationOnMock -> null);
+ TaskInstanceKillResponse response = taskInstanceKillOperationFunction.operate(
+ new TaskInstanceKillRequest(taskInstanceId));
+ Assertions.assertEquals("Cannot find WorkerTaskExecutor", response.getMessage());
+ }
+
+ try (MockedStatic logUtilsMockedStatic = Mockito.mockStatic(LogUtils.class)) {
+ int processId = 12;
+ int taskInstanceId = 111;
+ Mockito.reset(taskExecutionContext);
+ given(taskExecutionContext.getProcessId()).willReturn(processId);
+ given(taskExecutionContext.getLogPath()).willReturn("logpath");
+ logUtilsMockedStatic
+ .when(() -> LogUtils
+ .setTaskInstanceLogFullPathMDC(any(String.class)))
+ .then(invocationOnMock -> null);
+ taskInstanceKillOperationFunction.operate(
+ new TaskInstanceKillRequest(taskInstanceId));
+ logUtilsMockedStatic.verify(times(1), () -> LogUtils.removeTaskInstanceIdMDC());
+ logUtilsMockedStatic.verify(times(1), () -> LogUtils.removeTaskInstanceLogFullPathMDC());
+ }
+
+ try (MockedStatic logUtilsMockedStatic = Mockito.mockStatic(LogUtils.class)) {
+ try (
+ MockedStatic workerTaskExecutorHolderMockedStatic =
+ Mockito.mockStatic(WorkerTaskExecutorHolder.class)) {
+ given(workerTaskExecutor.getTaskExecutionContext()).willReturn(taskExecutionContext);
+ workerTaskExecutorHolderMockedStatic
+ .when(() -> WorkerTaskExecutorHolder.get(any(Integer.class)))
+ .thenReturn(workerTaskExecutor);
+ int processId = 12;
+ int taskInstanceId = 111;
+ Mockito.reset(taskExecutionContext);
+ given(taskExecutionContext.getProcessId()).willReturn(processId);
+ given(taskExecutionContext.getLogPath()).willReturn("logpath");
+ logUtilsMockedStatic
+ .when(() -> LogUtils
+ .setTaskInstanceLogFullPathMDC(any(String.class)))
+ .then(invocationOnMock -> null);
+ when(workerTaskExecutor.getTask()).thenReturn(task);
+ // given(workerManager.getTaskExecuteThread(taskInstanceId)).willReturn(workerTaskExecutor);
+ taskInstanceKillOperationFunction.operate(
+ new TaskInstanceKillRequest(taskInstanceId));
+ verify(task, times(1)).cancel();
+ }
+
+ }
+ }
+}
diff --git a/images/cncf-landscape-white-bg.jpg b/images/cncf-landscape-white-bg.jpg
new file mode 100644
index 0000000000..0ee426cbdf
Binary files /dev/null and b/images/cncf-landscape-white-bg.jpg differ
diff --git a/images/cncf-white-bg.jpg b/images/cncf-white-bg.jpg
new file mode 100644
index 0000000000..0e9579074f
Binary files /dev/null and b/images/cncf-white-bg.jpg differ
diff --git a/pom.xml b/pom.xml
index 260524b1e5..4a71741f8e 100755
--- a/pom.xml
+++ b/pom.xml
@@ -88,6 +88,7 @@
3.0.0
7.1.2
1.18.20
+ 4.2.0
apache
${project.name}
${project.version}
@@ -365,6 +366,12 @@
${lombok.version}
provided
+
+ org.awaitility
+ awaitility
+ ${awaitility.version}
+ test
+