Republish txn if error occured during publish

be/src/agent/task_worker_pool.cpp

@@ -812,9 +812,9 @@ void* TaskWorkerPool::_publish_version_worker_thread_callback(void* arg_this) {
 
         TFinishTaskRequest finish_task_request;
         if (res != OLAP_SUCCESS) {
-            // if publish failed, should also set status to ok, or fe will not deal with
-            // partial successfully tablet
-            status_code = TStatusCode::OK;
+            // if publish failed, return failed, fe will ignore this error and 
+            // check error tablet ids and fe will also republish this task
+            status_code = TStatusCode::RUNTIME_ERROR;
             LOG(WARNING) << "publish version failed. signature:" << agent_task_req.signature;
             error_msgs.push_back("publish version failed");
             finish_task_request.__set_error_tablet_ids(error_tablet_ids);

fe/src/main/java/org/apache/doris/master/MasterImpl.java

@@ -132,7 +132,7 @@ public TMasterResult finishTask(TFinishTaskRequest request) throws TException {
                 // We start to let FE perceive the task's error msg
                 if (taskType != TTaskType.MAKE_SNAPSHOT && taskType != TTaskType.UPLOAD
                         && taskType != TTaskType.DOWNLOAD && taskType != TTaskType.MOVE
-                        && taskType != TTaskType.CLONE) {
+                        && taskType != TTaskType.CLONE && taskType != TTaskType.PUBLISH_VERSION) {
                     return result;
                 }
             }
@@ -562,6 +562,10 @@ private void finishPublishVersion(AgentTask task, TFinishTaskRequest request) {
         PublishVersionTask publishVersionTask = (PublishVersionTask) task;
         publishVersionTask.addErrorTablets(errorTabletIds);
         publishVersionTask.setIsFinished(true);
+        if (request.getTask_status().getStatus_code() != TStatusCode.OK) {
+            // not remove the task from queue and be will retry
+            return;
+        }
         AgentTaskQueue.removeTask(publishVersionTask.getBackendId(), 
                                   publishVersionTask.getTaskType(), 
                                   publishVersionTask.getSignature());

fe/src/main/java/org/apache/doris/task/PublishVersionTask.java

@@ -58,11 +58,12 @@ public List<TPartitionVersionInfo> getPartitionVersionInfos() {
         return partitionVersionInfos;
     }
 
-    public List<Long> getErrorTablets() {
+    public synchronized List<Long> getErrorTablets() {
         return errorTablets;
     }
 
-    public void addErrorTablets(List<Long> errorTablets) {
+    public synchronized void addErrorTablets(List<Long> errorTablets) {
+        this.errorTablets.clear();
         if (errorTablets == null) {
             return;
         }