fix: unregister resource when closing DataSourceProxy

core/src/main/java/org/apache/seata/core/protocol/MessageType.java

@@ -138,6 +138,14 @@ public interface MessageType {
      * The constant TYPE_REG_RM_RESULT.
      */
     short TYPE_REG_RM_RESULT = 104;
+    /**
+     * The constant TYPE_UNREG_RM.
+     */
+    short TYPE_UNREG_RM = 105;
+    /**
+     * The constant TYPE_UNREG_RM_RESULT.
+     */
+    short TYPE_UNREG_RM_RESULT = 106;
     /**
      * The constant TYPE_RM_DELETE_UNDOLOG.
      */

core/src/main/java/org/apache/seata/core/protocol/UnRegisterRMRequest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.seata.core.protocol;
+
+import java.io.Serializable;
+
+/**
+ * The type UnRegistration rm request.
+ *
+ */
+public class UnRegisterRMRequest extends AbstractIdentifyRequest implements Serializable {
+
+    private static final long serialVersionUID = -4344171281014300302L;
+
+    /**
+     * Instantiates a new UnRegister rm request.
+     */
+    public UnRegisterRMRequest() {
+        this(null, null);
+    }
+
+    /**
+     * Instantiates a new UnRegister rm request.
+     *
+     * @param applicationId           the application id
+     * @param transactionServiceGroup the transaction service group
+     */
+    public UnRegisterRMRequest(String applicationId, String transactionServiceGroup) {
+        super(applicationId, transactionServiceGroup);
+    }
+
+    private String resourceIds;
+
+    /**
+     * Gets resource ids.
+     *
+     * @return the resource ids
+     */
+    public String getResourceIds() {
+        return resourceIds;
+    }
+
+    /**
+     * Sets resource ids.
+     *
+     * @param resourceIds the resource ids
+     */
+    public void setResourceIds(String resourceIds) {
+        this.resourceIds = resourceIds;
+    }
+
+    @Override
+    public short getTypeCode() {
+        return MessageType.TYPE_UNREG_RM;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("UnRegisterRMRequest{");
+        sb.append("resourceIds='").append(resourceIds).append('\'');
+        sb.append(", version='").append(version).append('\'');
+        sb.append(", applicationId='").append(applicationId).append('\'');
+        sb.append(", transactionServiceGroup='").append(transactionServiceGroup).append('\'');
+        sb.append(", extraData='").append(extraData).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+}

core/src/main/java/org/apache/seata/core/protocol/UnRegisterRMResponse.java

@@ -0,0 +1,51 @@
+/*
+ * 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.seata.core.protocol;
+
+import java.io.Serializable;
+
+/**
+ * The type UnRegister rm response.
+ *
+ */
+public class UnRegisterRMResponse extends AbstractIdentifyResponse implements Serializable {
+
+    private static final long serialVersionUID = -2824345339247012720L;
+
+    /**
+     * Instantiates a new UnRegister rm response.
+     */
+    public UnRegisterRMResponse() {
+        this(true);
+    }
+
+    /**
+     * Instantiates a new UnRegister rm response.
+     *
+     * @param result the result
+     */
+    public UnRegisterRMResponse(boolean result) {
+        super();
+        setIdentified(result);
+        setResultCode(result ? ResultCode.Success : ResultCode.Failed);
+    }
+
+    @Override
+    public short getTypeCode() {
+        return MessageType.TYPE_UNREG_RM_RESULT;
+    }
+}

core/src/main/java/org/apache/seata/core/rpc/RpcContext.java

@@ -91,6 +91,28 @@ public void release() {
         }
     }
 
+    /**
+     * Remove resources.
+     *
+     * @param resources the resources
+     */
+    public void removeResources(Set<String> resources) {
+        if (resources == null || resourceSets == null) {
+            return;
+        }
+        resourceSets.removeAll(resources);
+        if (clientRMHolderMap != null) {
+            Integer clientPort = ChannelUtil.getClientPortFromChannel(channel);
+            for (String resourceId : resources) {
+                Map<Integer, RpcContext> portMap = clientRMHolderMap.get(resourceId);
+                if (portMap != null) {
+                    portMap.remove(clientPort);
+                }
+                clientRMHolderMap.remove(resourceId);
+            }
+        }
+    }
+
     /**
      * Hold in client channels.
      *

core/src/main/java/org/apache/seata/core/rpc/netty/ChannelManager.java

@@ -25,6 +25,7 @@
 import org.apache.seata.core.protocol.IncompatibleVersionException;
 import org.apache.seata.core.protocol.RegisterRMRequest;
 import org.apache.seata.core.protocol.RegisterTMRequest;
+import org.apache.seata.core.protocol.UnRegisterRMRequest;
 import org.apache.seata.core.rpc.RpcContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -194,6 +195,39 @@ public static void registerRMChannel(RegisterRMRequest resourceManagerRequest, C
         }
     }
 
+    /**
+     * Unregister rm channel.
+     *
+     * @param unRegisterRMRequest the unregister rm request
+     * @param channel             the channel
+     */
+    public static void unregisterRMChannel(UnRegisterRMRequest unRegisterRMRequest, Channel channel) {
+        RpcContext rpcContext = IDENTIFIED_CHANNELS.get(channel);
+        if (rpcContext == null) {
+            return;
+        }
+        Set<String> dbkeySet = dbKeytoSet(unRegisterRMRequest.getResourceIds());
+        if (dbkeySet == null || dbkeySet.isEmpty()) {
+            return;
+        }
+        rpcContext.removeResources(dbkeySet);
+        for (String resourceId : dbkeySet) {
+            ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> applicationIdMap =
+                    RM_CHANNELS.get(resourceId);
+            if (applicationIdMap != null) {
+                ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> ipMap =
+                        applicationIdMap.get(unRegisterRMRequest.getApplicationId());
+                if (ipMap != null) {
+                    ConcurrentMap<Integer, RpcContext> portMap =
+                            ipMap.get(ChannelUtil.getClientIpFromChannel(channel));
+                    if (portMap != null) {
+                        portMap.remove(ChannelUtil.getClientPortFromChannel(channel));
+                    }
+                }
+            }
+        }
+    }
+
     private static void updateChannelsResource(String resourceId, String clientIp, String applicationId) {
         ConcurrentMap<Integer, RpcContext> sourcePortMap =
                 RM_CHANNELS.get(resourceId).get(applicationId).get(clientIp);

core/src/main/java/org/apache/seata/core/rpc/netty/NettyRemotingServer.java

@@ -23,6 +23,7 @@
 import org.apache.seata.core.rpc.TransactionMessageHandler;
 import org.apache.seata.core.rpc.processor.server.RegRmProcessor;
 import org.apache.seata.core.rpc.processor.server.RegTmProcessor;
+import org.apache.seata.core.rpc.processor.server.UnRegRmProcessor;
 import org.apache.seata.core.rpc.processor.server.ServerHeartbeatProcessor;
 import org.apache.seata.core.rpc.processor.server.ServerOnRequestProcessor;
 import org.apache.seata.core.rpc.processor.server.ServerOnResponseProcessor;
@@ -121,6 +122,8 @@ private void registerProcessor() {
         // 3. registry rm message processor
         RegRmProcessor regRmProcessor = new RegRmProcessor(this);
         super.registerProcessor(MessageType.TYPE_REG_RM, regRmProcessor, messageExecutor);
+        UnRegRmProcessor unRegRmProcessor = new UnRegRmProcessor(this);
+        super.registerProcessor(MessageType.TYPE_UNREG_RM, unRegRmProcessor, messageExecutor);
         // 4. registry tm message processor
         RegTmProcessor regTmProcessor = new RegTmProcessor(this);
         super.registerProcessor(MessageType.TYPE_REG_CLT, regTmProcessor, null);

core/src/main/java/org/apache/seata/core/rpc/netty/RmNettyRemotingClient.java

@@ -33,6 +33,8 @@
 import org.apache.seata.core.protocol.MessageType;
 import org.apache.seata.core.protocol.RegisterRMRequest;
 import org.apache.seata.core.protocol.RegisterRMResponse;
+import org.apache.seata.core.protocol.UnRegisterRMRequest;
+import org.apache.seata.core.protocol.Version;
 import org.apache.seata.core.rpc.netty.NettyPoolKey.TransactionRole;
 import org.apache.seata.core.rpc.processor.client.ClientHeartbeatProcessor;
 import org.apache.seata.core.rpc.processor.client.ClientOnResponseProcessor;
@@ -275,6 +277,46 @@ public void registerResource(String resourceGroupId, String resourceId) {
         }
     }
 
+    /**
+     * Unregister resource.
+     *
+     * @param resourceGroupId the resource group id
+     * @param resourceId      the resource id
+     */
+    public void unRegisterResource(String resourceGroupId, String resourceId) {
+        if (StringUtils.isBlank(transactionServiceGroup) || StringUtils.isBlank(resourceId)) {
+            return;
+        }
+
+        if (getClientChannelManager().getChannels().isEmpty()) {
+            return;
+        }
+
+        synchronized (getClientChannelManager().getChannels()) {
+            for (Map.Entry<String, Channel> entry : getClientChannelManager().getChannels().entrySet()) {
+                Channel rmChannel = entry.getValue();
+                String serverVersion = getClientChannelManager().getChannelVersion(rmChannel);
+                // Threshold for version-aware unregistration: version >= 2.1.0
+                if (Version.isAboveOrEqualVersion(serverVersion, "2.1.0")) {
+                    if (LOGGER.isInfoEnabled()) {
+                        LOGGER.info("will unregister resourceId:{}", resourceId);
+                    }
+                    sendUnRegisterMessage(rmChannel, resourceId);
+                }
+            }
+        }
+    }
+
+    private void sendUnRegisterMessage(Channel channel, String resourceId) {
+        UnRegisterRMRequest message = new UnRegisterRMRequest(applicationId, transactionServiceGroup);
+        message.setResourceIds(resourceId);
+        try {
+            super.sendAsyncRequest(channel, message);
+        } catch (FrameworkException e) {
+            LOGGER.error("unregister resource failed, channel:{},resourceId:{}", channel, resourceId, e);
+        }
+    }
+
     public void sendRegisterMessage(String serverAddress, Channel channel, String resourceId) {
         RegisterRMRequest message = new RegisterRMRequest(applicationId, transactionServiceGroup);
         message.setResourceIds(resourceId);
@@ -371,6 +413,7 @@ private void registerProcessor() {
         super.registerProcessor(MessageType.TYPE_BRANCH_STATUS_REPORT_RESULT, onResponseProcessor, null);
         super.registerProcessor(MessageType.TYPE_GLOBAL_LOCK_QUERY_RESULT, onResponseProcessor, null);
         super.registerProcessor(MessageType.TYPE_REG_RM_RESULT, onResponseProcessor, null);
+        super.registerProcessor(MessageType.TYPE_UNREG_RM_RESULT, onResponseProcessor, null);
         super.registerProcessor(MessageType.TYPE_BATCH_RESULT_MSG, onResponseProcessor, null);
         // 5.registry heartbeat message processor
         ClientHeartbeatProcessor clientHeartbeatProcessor = new ClientHeartbeatProcessor();

core/src/main/java/org/apache/seata/core/rpc/processor/server/UnRegRmProcessor.java

@@ -0,0 +1,67 @@
+/*
+ * 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.seata.core.rpc.processor.server;
+
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.seata.core.protocol.RpcMessage;
+import org.apache.seata.core.protocol.UnRegisterRMRequest;
+import org.apache.seata.core.protocol.UnRegisterRMResponse;
+import org.apache.seata.core.rpc.RemotingServer;
+import org.apache.seata.core.rpc.netty.ChannelManager;
+import org.apache.seata.core.rpc.processor.RemotingProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * process RM client unRegistration message.
+ * <p>
+ * process message type:
+ * {@link UnRegisterRMRequest}
+ *
+ * @since 2.x
+ */
+public class UnRegRmProcessor implements RemotingProcessor {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(UnRegRmProcessor.class);
+
+    private RemotingServer remotingServer;
+
+    public UnRegRmProcessor(RemotingServer remotingServer) {
+        this.remotingServer = remotingServer;
+    }
+
+    @Override
+    public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
+        onUnRegRmMessage(ctx, rpcMessage);
+    }
+
+    private void onUnRegRmMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
+        UnRegisterRMRequest message = (UnRegisterRMRequest) rpcMessage.getBody();
+        boolean isSuccess = true;
+        try {
+            ChannelManager.unregisterRMChannel(message, ctx.channel());
+        } catch (Exception exx) {
+            isSuccess = false;
+            LOGGER.error("RM unregister fail, error message:{}", exx.getMessage());
+        }
+        UnRegisterRMResponse response = new UnRegisterRMResponse(isSuccess);
+        remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), response);
+        if (isSuccess && LOGGER.isInfoEnabled()) {
+            LOGGER.info("RM unregister success, message:{}, channel:{}", message, ctx.channel());
+        }
+    }
+}

rm-datasource/src/main/java/org/apache/seata/rm/datasource/DataSourceManager.java

@@ -93,7 +93,9 @@ public void registerResource(Resource resource) {
 
     @Override
     public void unregisterResource(Resource resource) {
-        throw new NotSupportYetException("unregister a resource");
+        DataSourceProxy dataSourceProxy = (DataSourceProxy) resource;
+        dataSourceCache.remove(dataSourceProxy.getResourceId());
+        super.unregisterResource(dataSourceProxy);
     }
 
     /**

rm-datasource/src/main/java/org/apache/seata/rm/datasource/DataSourceProxy.java

@@ -294,7 +294,7 @@ public String getUserName() {
     }
 
     public void close() throws Exception {
-        // TODO: Need to unregister resource from DefaultResourceManager
+        DefaultResourceManager.get().unregisterResource(this);
         TableMetaCacheFactory.shutdown(resourceId);
     }
 }