Apache DolphinScheduler 从 1.3.4 升级至3.1.2 过程中的问题记录及解决方案

<property>    <name>dfs.namenode.kerberos.principal.patternname>    <value>*value>property>

update t_ds_task_instance set log_path=replace(log_path,'/logs/','/worker-server/logs/');

cp -r {旧版本dolphinscheduler目录}/logs/[1-9]* {新版本dolphinscheduler目录}/worker-server/logs/*

# 查出主键自增值select AUTO_INCREMENT FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'dolphinscheduler' AND TABLE_NAME = 't_ds_process_definition' limit 1# 将上面sql的执行结果填写到下方参数处执行alter table dolphinscheduler_bak1.t_ds_process_definition_log auto_increment = {max_id};

  select        <include refid="baseSqlV2">            <property name="alias" value="instance"/>        include>        ,        process.name as process_instance_name        from t_ds_task_instance instance        left join t_ds_task_definition_log define on define.code=instance.task_code and define.version=instance.task_definition_version        left join t_ds_process_instance process on process.id=instance.process_instance_id        where define.project_code = #{projectCode}        <if test="startTime != null">            and instance.start_time =]]> #{startTime}        if>  ......省略多余部分

     select        <include refid="baseSqlV2">            <property name="alias" value="instance"/>        include>        ,        process.name as process_instance_name        from t_ds_task_instance instance--         left join t_ds_task_definition_log define --    on define.code=instance.task_code and --     define.version=instance.task_definition_version        join t_ds_process_instance process         on process.id=instance.process_instance_id        join t_ds_process_definition define         on define.code=process.process_definition_code        where define.project_code = #{projectCode}        <if test="startTime != null">            and instance.start_time =]]> #{startTime}        if>  ......省略多余部分

513 if (TASK_TYPE_SUB_PROCESS.equals(taskType)) {514                       JsonNode jsonNodeDefinitionId = param.get("processDefinitionId");515                       if (jsonNodeDefinitionId != null) {516                           param.put("processDefinitionCode",517                                  processDefinitionMap.get(jsonNodeDefinitionId.asInt()).getCode());518                            param.remove("processDefinitionId");519                        }520                    }

if (jsonNodeDefinitionId != null) {    if (processDefinitionMap.get(jsonNodeDefinitionId.asInt()) != null) {        param.put("processDefinitionCode",processDefinitionMap.get(jsonNodeDefinitionId.asInt()).getCode());        param.remove("processDefinitionId");    } else {        logger.error("*******************error");        logger.error("*******************param:" + param);        logger.error("*******************jsonNodeDefinitionId:" + jsonNodeDefinitionId);    }}

669 if (mapEntry.isPresent()) {670                            Map.Entry<long, map> processCodeTaskNameCodeEntry = mapEntry.get();671                            dependItem.put("definitionCode", processCodeTaskNameCodeEntry.getKey());672                            String depTasks = dependItem.get("depTasks").asText();673                            long taskCode =674                                    "ALL".equals(depTasks) || processCodeTaskNameCodeEntry.getValue() == null ? 0L675                                            : processCodeTaskNameCodeEntry.getValue().get(depTasks);676                            dependItem.put("depTaskCode", taskCode);677                        }

long taskCode =0;                            if (processCodeTaskNameCodeEntry.getValue() != null                                    &&processCodeTaskNameCodeEntry.getValue().get(depTasks)!=null){                                taskCode =processCodeTaskNameCodeEntry.getValue().get(depTasks);                            }else{                                logger.error("******************** depTasks:"+depTasks);                                logger.error("******************** taskCode not in "+JSONUtils.toJsonString(processCodeTaskNameCodeEntry));                            }                            dependItem.put("depTaskCode", taskCode);

security:  authentication:    # Authentication types (supported types: PASSWORD,LDAP)    type: LDAP    # IF you set type `LDAP`, below config will be effective    ldap:      # ldap server config      urls: xxx      base-dn: xxx      username: xxx      password: xxx      user:        # admin userId when you use LDAP login        admin: xxx        identity-attribute: xxx        email-attribute: xxx        # action when ldap user is not exist (supported types: CREATE,DENY)        not-exist-action: CREATE

ctx = new InitialLdapContext(searchEnv, null);SearchControls sc = new SearchControls();sc.setReturningAttributes(new String[]{ldapEmailAttribute});sc.setSearchScope(SearchControls.SUBTREE_SCOPE);EqualsFilter filter = new EqualsFilter(ldapUserIdentifyingAttribute, userId);NamingEnumeration results = ctx.search(ldapBaseDn, filter.toString(), sc);if (results.hasMore()) {    // get the users DN (distinguishedName) from the result    SearchResult result = results.next();    NamingEnumeration attrs = result.getAttributes().getAll();    while (attrs.hasMore()) {        // Open another connection to the LDAP server with the found DN and the password        searchEnv.put(Context.SECURITY_PRINCIPAL, result.getNameInNamespace());        searchEnv.put(Context.SECURITY_CREDENTIALS, userPwd);        try {            new InitialDirContext(searchEnv);        } catch (Exception e) {            logger.warn("invalid ldap credentials or ldap search error", e);            return null;        }        Attribute attr = attrs.next();        if (attr.getID().equals(ldapEmailAttribute)) {            return (String) attr.get();        }    }}

// sc.setReturningAttributes(new String[]{ldapEmailAttribute});

NamingEnumeration attrs = result.getAttributes().getAll();

@Override        public Set listAuthorizedResource(int userId, Logger logger) {            List relationResources;            if (userId == 0) {                relationResources = new ArrayList<>();            } else {                // query resource relation                List resIds = resourceUserMapper.queryResourcesIdListByUserIdAndPerm(userId, 0);                relationResources = CollectionUtils.isEmpty(resIds) ? new ArrayList<>() : resourceMapper.queryResourceListById(resIds);            }            List ownResourceList = resourceMapper.queryResourceListAuthored(userId, -1);            relationResources.addAll(ownResourceList);            return relationResources.stream().map(Resource::getId).collect(toSet()); // 解决资源文件授权无效的问题//            return ownResourceList.stream().map(Resource::getId).collect(toSet());        }

 /**     * * 定时更新凭证     */    private static void startCheckKeytabTgtAndReloginJob() {        // 每天循环，定时更新凭证        Executors.newScheduledThreadPool(1).scheduleWithFixedDelay(() -> {            try {                UserGroupInformation.getLoginUser().checkTGTAndReloginFromKeytab();                logger.warn("Check Kerberos Tgt And Relogin From Keytab Finish.");            } catch (IOException e) {                logger.error("Check Kerberos Tgt And Relogin From Keytab Error", e);            }        }, 0, 1, TimeUnit.DAYS);        logger.info("Start Check Keytab TGT And Relogin Job Success.");    }

public static boolean loadKerberosConf(String javaSecurityKrb5Conf, String loginUserKeytabUsername,                                           String loginUserKeytabPath, Configuration configuration) throws IOException {        if (CommonUtils.getKerberosStartupState()) {            System.setProperty(Constants.JAVA_SECURITY_KRB5_CONF, StringUtils.defaultIfBlank(javaSecurityKrb5Conf,                    PropertyUtils.getString(Constants.JAVA_SECURITY_KRB5_CONF_PATH)));            configuration.set(Constants.HADOOP_SECURITY_AUTHENTICATION, Constants.KERBEROS);            UserGroupInformation.setConfiguration(configuration);            UserGroupInformation.loginUserFromKeytab(                    StringUtils.defaultIfBlank(loginUserKeytabUsername,                            PropertyUtils.getString(Constants.LOGIN_USER_KEY_TAB_USERNAME)),                    StringUtils.defaultIfBlank(loginUserKeytabPath,                            PropertyUtils.getString(Constants.LOGIN_USER_KEY_TAB_PATH)));            startCheckKeytabTgtAndReloginJob();  // 此处调用            return true;        }        return false;    }