From 09adcf39126c025bcd09e61341ae6c71f2216d6b Mon Sep 17 00:00:00 2001
From: zhumingze <zhumingze@oppo.com>
Date: Mon, 17 Nov 2025 19:03:39 +0800
Subject: [PATCH] fix(master): logic adapts for decommission parts of the dp on
 the node. #1000441588

Signed-off-by: zhumingze <zhumingze@oppo.com>
---
 master/cluster.go      | 22 +++++++++++++++++++++-
 master/data_node.go    |  8 ++++++--
 master/disk_manager.go |  8 +-------
 3 files changed, 28 insertions(+), 10 deletions(-)

diff --git a/master/cluster.go b/master/cluster.go
index cdb08af61..17d8e147a 100644
--- a/master/cluster.go
+++ b/master/cluster.go
@@ -4792,6 +4792,7 @@ func (c *Cluster) checkDecommissionDataNode() {
 				//	dataNode.resetDecommissionStatus()
 				//	c.syncUpdateDataNode(dataNode)
 				// }
+
 				return true
 			}
 			// maybe has decommission failed dp
@@ -5190,10 +5191,29 @@ func (c *Cluster) checkDecommissionDisk() {
 	c.DecommissionDisks.Range(func(key, value interface{}) bool {
 		disk := value.(*DecommissionDisk)
 		status := disk.GetDecommissionStatus()
+		dataNode, err := c.dataNode(disk.SrcAddr)
+		if err != nil {
+			if strings.Contains(err.Error(), "not found") {
+				c.DecommissionDisks.Delete(key)
+				c.syncDeleteDecommissionDisk(disk)
+			}
+			return true
+		}
+		if status == DecommissionSuccess || status == DecommissionFail {
+			partitions := dataNode.badPartitions(disk.DiskPath, c, true)
+			// if only decommission part of data partitions, do not add disk to success list
+			if len(partitions) != 0 {
+				if dataNode.DecommissionLimit != 0 && !dataNode.isBadDisk(disk.DiskPath) {
+					// can allocate dp again
+					c.deleteAndSyncDecommissionedDisk(dataNode, disk.DiskPath)
+				}
+				return true
+			}
+		}
 		// keep failed decommission disk in list for preventing the reuse of a
 		// term in future decommissioning operations
 		if status == DecommissionSuccess {
-			c.addAndSyncDecommissionSuccessDisk(disk.SrcAddr, disk.DiskPath)
+			c.addAndSyncDecommissionSuccessDisk(dataNode, disk.DiskPath)
 			if time.Since(time.Unix(disk.DecommissionCompleteTime, 0)) > (120 * time.Hour) {
 				if err := c.syncDeleteDecommissionDisk(disk); err != nil {
 					msg := fmt.Sprintf("action[checkDecommissionDisk],clusterID[%v] node[%v] disk[%v],"+
diff --git a/master/data_node.go b/master/data_node.go
index 113de9ba5..de633a46d 100644
--- a/master/data_node.go
+++ b/master/data_node.go
@@ -638,7 +638,7 @@ func (dataNode *DataNode) updateDecommissionStatus(c *Cluster, debug, persist bo
 	if successDiskNum+failedDiskNum+cancelDiskNum == totalDisk {
 		if successDiskNum == totalDisk {
 			if persist {
-				dataNode.SetDecommissionStatus(DecommissionSuccess)
+				dataNode.markDecommissionSuccess(c)
 			}
 			return DecommissionSuccess, float64(1)
 		}
@@ -650,7 +650,7 @@ func (dataNode *DataNode) updateDecommissionStatus(c *Cluster, debug, persist bo
 			}
 		} else {
 			if persist {
-				dataNode.SetDecommissionStatus(DecommissionFail)
+				dataNode.markDecommissionFail()
 			} else {
 				return DecommissionFail, progress / float64(totalDisk)
 			}
@@ -750,6 +750,10 @@ func (dataNode *DataNode) markDecommissionSuccess(c *Cluster) {
 
 func (dataNode *DataNode) markDecommissionFail() {
 	dataNode.SetDecommissionStatus(DecommissionFail)
+	// if only decommission part of data partitions, can alloc dp in future
+	if dataNode.DecommissionLimit != 0 {
+		dataNode.ToBeOffline = false
+	}
 	// dataNode.ToBeOffline = false
 	// dataNode.DecommissionCompleteTime = time.Now().Unix()
 }
diff --git a/master/disk_manager.go b/master/disk_manager.go
index e295e8857..0aa5ba6b9 100644
--- a/master/disk_manager.go
+++ b/master/disk_manager.go
@@ -277,13 +277,7 @@ func (c *Cluster) deleteAndSyncDecommissionedDisk(dataNode *DataNode, diskPath s
 	return
 }
 
-func (c *Cluster) addAndSyncDecommissionSuccessDisk(addr string, diskPath string) (err error) {
-	var dataNode *DataNode
-	if dataNode, err = c.dataNode(addr); err != nil {
-		log.LogWarnf("action[addAndSyncDecommissionSuccessDisk] cannot find dataNode[%s]", addr)
-		return
-	}
-
+func (c *Cluster) addAndSyncDecommissionSuccessDisk(dataNode *DataNode, diskPath string) (err error) {
 	if exist := dataNode.addDecommissionSuccessDisk(diskPath); exist {
 		return
 	}