Add backoff in TransferQueueProcessor readTransferTasks (cadence-work…

…flow#332) Issue cadence-workflow#330
mudit3774 · Aug 24, 2017 · 37a6a24 · 37a6a24
1 parent df9ac3e
commit 37a6a24
Showing 1 changed file with 12 additions and 5 deletions.
diff --git a/service/history/transferQueueProcessor.go b/service/history/transferQueueProcessor.go
@@ -849,12 +849,19 @@ func (a *ackManager) readTransferTasks() ([]*persistence.TransferTaskInfo, error
 	a.RLock()
 	rLevel := a.readLevel
 	a.RUnlock()
-	response, err := a.executionMgr.GetTransferTasks(&persistence.GetTransferTasksRequest{
-		ReadLevel:    rLevel,
-		MaxReadLevel: a.shard.GetTransferMaxReadLevel(),
-		BatchSize:    a.processor.config.TransferTaskBatchSize,
-	})
 
+	var response *persistence.GetTransferTasksResponse
+	op := func() error {
+		var err error
+		response, err = a.executionMgr.GetTransferTasks(&persistence.GetTransferTasksRequest{
+			ReadLevel:    rLevel,
+			MaxReadLevel: a.shard.GetTransferMaxReadLevel(),
+			BatchSize:    a.processor.config.TransferTaskBatchSize,
+		})
+		return err
+	}
+
+	err := backoff.Retry(op, persistenceOperationRetryPolicy, common.IsPersistenceTransientError)
 	if err != nil {
 		return nil, err
 	}