@Override
    public ListenerResult listen(WorkflowContext context) throws Exception {
        ProcessForm processForm = context.getProcessForm();
        String groupId = processForm.getInlongGroupId();
        if (!(processForm instanceof GroupResourceProcessForm)) {
            String message = String.format("process form was not GroupResourceProcessForm for groupId=%s", groupId);
            log.error(message);
            return ListenerResult.fail(message);
        }

        GroupResourceProcessForm groupResourceProcessForm = (GroupResourceProcessForm) processForm;
        List<InlongStreamInfo> streamInfos = groupResourceProcessForm.getStreamInfos();
        for (InlongStreamInfo streamInfo : streamInfos) {
            List<StreamSink> sinkList = streamInfo.getSinkList();
            if (CollectionUtils.isEmpty(sinkList)) {
                continue;
            }
            List<InlongStreamExtInfo> extList = streamInfo.getExtList();
            log.info("stream sink ext info: {}", extList);

            Map<String, String> kvConf = new HashMap<>();
            extList.forEach(v -> kvConf.put(v.getKeyName(), v.getKeyValue()));
            String sortExt = kvConf.get(InlongConstants.SORT_PROPERTIES);
            if (StringUtils.isNotEmpty(sortExt)) {
                Map<String, String> result = JsonUtils.OBJECT_MAPPER.convertValue(
                        JsonUtils.OBJECT_MAPPER.readTree(sortExt), new TypeReference<Map<String, String>>() {
                        });
                kvConf.putAll(result);
            }

            String jobId = kvConf.get(InlongConstants.SORT_JOB_ID);
            if (StringUtils.isBlank(jobId)) {
                String message = String.format("sort job id is empty for groupId=%s, streamId=%s", groupId,
                        streamInfo.getInlongStreamId());
                return ListenerResult.fail(message);
            }

            FlinkInfo flinkInfo = new FlinkInfo();
            flinkInfo.setJobId(jobId);
            String sortUrl = kvConf.get(InlongConstants.SORT_URL);
            flinkInfo.setEndpoint(sortUrl);

            FlinkOperation flinkOperation = FlinkOperation.getInstance();
            try {
                flinkOperation.delete(flinkInfo);
                // TODO if the job is OFFLINE, should delete the scheduler information
                log.info("job delete success for jobId={}", jobId);
            } catch (Exception e) {
                flinkInfo.setException(true);
                flinkInfo.setExceptionMsg(getExceptionStackMsg(e));
                flinkOperation.pollJobStatus(flinkInfo, JobStatus.CANCELED);

                String message = String.format("delete sort failed for groupId=%s, streamId=%s", groupId,
                        streamInfo.getInlongStreamId());
                log.error(message, e);
                return ListenerResult.fail(message + ": " + e.getMessage());
            }
            flinkOperation.pollJobStatus(flinkInfo, JobStatus.CANCELED);
        }
        return ListenerResult.success();
    }
Listener of delete Sort task or config.
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java