Merge pull request #565 from didi/dev

合并开发分支

Author: ZQKC

docs/dev_guide/登录系统对接.md

@@ -0,0 +1,199 @@
+
+
+![Logo](https://user-images.githubusercontent.com/71620349/185368586-aed82d30-1534-453d-86ff-ecfa9d0f35bd.png)
+
+## 登录系统对接
+
+[KnowStreaming](https://github.com/didi/KnowStreaming)（以下简称KS） 除了实现基于本地MySQL的用户登录认证方式外，还已经实现了基于Ldap的登录认证。
+
+但是，登录认证系统并非仅此两种。因此，为了具有更好的拓展性，KS具有自定义登陆认证逻辑，快速对接已有系统的特性。
+
+在KS中，我们将登陆认证相关的一些文件放在[km-extends](https://github.com/didi/KnowStreaming/tree/master/km-extends)模块下的[km-account](https://github.com/didi/KnowStreaming/tree/master/km-extends/km-account)模块里。
+
+本文将介绍KS如何快速对接自有的用户登录认证系统。
+
+### 对接步骤
+
+- 创建一个登陆认证类，实现[LogiCommon](https://github.com/didi/LogiCommon)的LoginExtend接口；
+- 将[application.yml](https://github.com/didi/KnowStreaming/blob/master/km-rest/src/main/resources/application.yml)中的spring.logi-security.login-extend-bean-name字段改为登陆认证类的bean名称；
+
+```Java
+//LoginExtend 接口
+public interface LoginExtend {
+  
+    /**
+     * 验证登录信息，同时记住登录状态
+     */
+    UserBriefVO verifyLogin(AccountLoginDTO var1, HttpServletRequest var2, HttpServletResponse var3) throws LogiSecurityException;
+  
+    /**
+     * 登出接口，清楚登录状态
+     */
+    Result<Boolean> logout(HttpServletRequest var1, HttpServletResponse var2);
+  
+    /**
+     * 检查是否已经登录
+     */
+    boolean interceptorCheck(HttpServletRequest var1, HttpServletResponse var2, String var3, List<String> var4) throws IOException;
+  
+}
+
+```
+
+
+
+### 对接例子
+
+我们以Ldap对接为例，说明KS如何对接登录认证系统。
+
++ 编写[LdapLoginServiceImpl](https://github.com/didi/KnowStreaming/blob/master/km-extends/km-account/src/main/java/com/xiaojukeji/know/streaming/km/account/login/ldap/LdapLoginServiceImpl.java)类，实现LoginExtend接口。
++ 设置[application.yml](https://github.com/didi/KnowStreaming/blob/master/km-rest/src/main/resources/application.yml)中的spring.logi-security.login-extend-bean-name=ksLdapLoginService。
+
+完成上述两步即可实现KS对接Ldap认证登陆。
+
+```Java
+@Service("ksLdapLoginService")
+public class LdapLoginServiceImpl implements LoginExtend  {
+  
+
+    @Override
+    public UserBriefVO verifyLogin(AccountLoginDTO loginDTO,
+                                   HttpServletRequest request,
+                                   HttpServletResponse response) throws LogiSecurityException {
+        String decodePasswd = AESUtils.decrypt(loginDTO.getPw());
+
+        // 去LDAP验证账密
+        LdapPrincipal ldapAttrsInfo = ldapAuthentication.authenticate(loginDTO.getUserName(), decodePasswd);
+        if (ldapAttrsInfo == null) {
+            // 用户不存在，正常来说上如果有问题，上一步会直接抛出异常
+            throw new LogiSecurityException(ResultCode.USER_NOT_EXISTS);
+        }
+
+        // 进行业务相关操作
+        
+        // 记录登录状态，Ldap因为无法记录登录状态，因此有KnowStreaming进行记录
+        initLoginContext(request, response, loginDTO.getUserName(), user.getId());
+        return CopyBeanUtil.copy(user, UserBriefVO.class);
+    }
+
+    @Override
+    public Result<Boolean> logout(HttpServletRequest request, HttpServletResponse response) {
+        
+        //清理cookie和session
+      
+        return Result.buildSucc(Boolean.TRUE);
+    }
+
+    @Override
+    public boolean interceptorCheck(HttpServletRequest request, HttpServletResponse response, String requestMappingValue, List<String> whiteMappingValues) throws IOException {
+       
+        // 检查是否已经登录
+        String userName = HttpRequestUtil.getOperator(request);
+        if (StringUtils.isEmpty(userName)) {
+            // 未登录，则进行登出
+            logout(request, response);
+            return Boolean.FALSE;
+        }
+
+        return Boolean.TRUE;
+    }
+}
+
+```
+
+
+
+### 实现原理
+
+因为登陆和登出整体实现逻辑是一致的，所以我们以登陆逻辑为例进行介绍。
+
++ 登陆原理
+
+登陆走的是[LogiCommon](https://github.com/didi/LogiCommon)自带的LoginController。
+
+```java
+@RestController
+public class LoginController {
+  
+
+    //登陆接口
+    @PostMapping({"/login"})
+    public Result<UserBriefVO> login(HttpServletRequest request, HttpServletResponse response, @RequestBody AccountLoginDTO loginDTO) {
+        try {
+             //登陆认证
+            UserBriefVO userBriefVO = this.loginService.verifyLogin(loginDTO, request, response);
+            return Result.success(userBriefVO);
+          
+        } catch (LogiSecurityException var5) {
+            return Result.fail(var5);
+        }
+    }
+
+}
+```
+
+而登陆操作是调用LoginServiceImpl类来实现，但是具体由哪个登陆认证类来执行登陆操作却由loginExtendBeanTool来指定。
+
+```java
+//LoginServiceImpl类
+@Service
+public class LoginServiceImpl implements LoginService {
+
+    //实现登陆操作，但是具体哪个登陆类由loginExtendBeanTool来管理
+    public UserBriefVO verifyLogin(AccountLoginDTO loginDTO, HttpServletRequest request, HttpServletResponse response) throws LogiSecurityException {
+      
+        return this.loginExtendBeanTool.getLoginExtendImpl().verifyLogin(loginDTO, request, response);
+    }
+
+
+}
+```
+
+而loginExtendBeanTool类会优先去查找用户指定的登陆认证类，如果失败则调用默认的登陆认证函数。
+
+```java
+//LoginExtendBeanTool类 
+@Component("logiSecurityLoginExtendBeanTool")
+public class LoginExtendBeanTool {
+
+    public LoginExtend getLoginExtendImpl() {
+        LoginExtend loginExtend;
+      	//先调用用户指定登陆类，如果失败则调用系统默认登陆认证
+        try {
+          //调用的类由spring.logi-security.login-extend-bean-name指定
+            loginExtend = this.getCustomLoginExtendImplBean();
+        } catch (UnsupportedOperationException var3) {
+            loginExtend = this.getDefaultLoginExtendImplBean();
+        }
+
+        return loginExtend;
+    }
+}
+```
+
++ 认证原理
+
+认证的实现则比较简单，向Spring中注册我们的拦截器PermissionInterceptor。
+
+拦截器会调用LoginServiceImpl类的拦截方法，LoginServiceImpl后续处理逻辑就和前面登陆是一致的。
+
+```java
+public class PermissionInterceptor implements HandlerInterceptor {
+
+
+    /**
+     * 拦截预处理
+     * @return boolean false:拦截, 不向下执行, true:放行
+     */
+    @Override
+    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
+       
+      	//免登录相关校验，如果验证通过，提前返回
+				
+      	//走拦截函数，进行普通用户验证
+        return loginService.interceptorCheck(request, response, classRequestMappingValue, whiteMappingValues);
+    }
+
+}
+```
+

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/group/impl/GroupManagerImpl.java

@@ -20,6 +20,7 @@
 import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
 import com.xiaojukeji.know.streaming.km.common.enums.AggTypeEnum;
 import com.xiaojukeji.know.streaming.km.common.enums.GroupOffsetResetEnum;
+import com.xiaojukeji.know.streaming.km.common.enums.group.GroupStateEnum;
 import com.xiaojukeji.know.streaming.km.common.exception.AdminOperateException;
 import com.xiaojukeji.know.streaming.km.common.exception.NotExistException;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -171,7 +172,7 @@ public Result<Void> resetGroupOffsets(GroupOffsetResetDTO dto, String operator)
         }
 
         if (!ConsumerGroupState.EMPTY.equals(description.state()) && !ConsumerGroupState.DEAD.equals(description.state())) {
-            return Result.buildFromRSAndMsg(ResultStatus.KAFKA_OPERATE_FAILED, String.format("group处于%s, 重置失败(仅Empty情况可重置)", description.state().name()));
+            return Result.buildFromRSAndMsg(ResultStatus.KAFKA_OPERATE_FAILED, String.format("group处于%s, 重置失败(仅Empty情况可重置)", GroupStateEnum.getByRawState(description.state()).getState()));
         }
 
         // 获取offset

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/event/cluster/ClusterPhyAddedEvent.java

@@ -0,0 +1,15 @@
+package com.xiaojukeji.know.streaming.km.common.bean.event.cluster;
+
+import lombok.Getter;
+
+/**
+ * 集群新增事件
+ * @author zengqiao
+ * @date 22/02/25
+ */
+@Getter
+public class ClusterPhyAddedEvent extends ClusterPhyBaseEvent {
+    public ClusterPhyAddedEvent(Object source, Long clusterPhyId) {
+        super(source, clusterPhyId);
+    }
+}

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/event/kafka/zk/BaseKafkaZKEvent.java

@@ -170,6 +170,7 @@ private static ReassignJobDetailDataGroupByTopic convert2ReassignJobDetailDataGr
             detail.setOriginalBrokerIdList(CommonUtils.string2IntList(subJobPO.getOriginalBrokerIds()));
             detail.setReassignBrokerIdList(CommonUtils.string2IntList(subJobPO.getReassignBrokerIds()));
             detail.setStatus(subJobPO.getStatus());
+            detail.setOldReplicaNum(detail.getOriginalBrokerIdList().size());
 
             ReassignSubJobExtendData extendData = ConvertUtil.str2ObjByJson(subJobPO.getExtendData(), ReassignSubJobExtendData.class);
             if (extendData != null) {
@@ -217,6 +218,7 @@ private static ReassignJobDetailDataGroupByTopic convert2ReassignJobDetailDataGr
 
         topicDetail.setPresentReplicaNum(partitionDetailList.get(0).getPresentReplicaNum());
         topicDetail.setNewReplicaNum(partitionDetailList.get(0).getNewReplicaNum());
+        topicDetail.setOldReplicaNum(partitionDetailList.get(0).getOldReplicaNum());
         topicDetail.setOriginalRetentionTimeUnitMs(partitionDetailList.get(0).getOriginalRetentionTimeUnitMs());
         topicDetail.setReassignRetentionTimeUnitMs(partitionDetailList.get(0).getReassignRetentionTimeUnitMs());
 

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/event/kafka/zk/ControllerChangeEvent.java

@@ -7,7 +7,9 @@
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
+import com.xiaojukeji.know.streaming.km.common.bean.event.cluster.ClusterPhyAddedEvent;
 import com.xiaojukeji.know.streaming.km.common.bean.po.cluster.ClusterPhyPO;
+import com.xiaojukeji.know.streaming.km.common.component.SpringTool;
 import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
 import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.ModuleEnum;
 import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.OperationEnum;
@@ -106,6 +108,8 @@ public Long addClusterPhy(ClusterPhyPO clusterPhyPO, String operator) throws Par
 
             log.info("method=addClusterPhy||clusterPhyId={}||operator={}||msg=add cluster finished", clusterPhyPO.getId(), operator);
 
+            // 发布添加集群事件
+            SpringTool.publish(new ClusterPhyAddedEvent(this, clusterPhyPO.getId()));
             return clusterPhyPO.getId();
         } catch (DuplicateKeyException dke) {
             log.warn("method=addClusterPhy||clusterPhyId={}||operator={}||msg=add cluster failed||errMsg=duplicate data", clusterPhyPO.getId(), operator);

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/converter/ReassignConverter.java

@@ -60,4 +60,7 @@ public interface ReassignJobService {
      * 依据任务状态或者其中一个任务ID
      */
     Long getOneRunningJobId(Long clusterPhyId);
+
+
+    Result<Void> preferredReplicaElection(Long jobId);
 }