remove idl changes (#3444)

cadence-workflow · Aug 11, 2020 · 68a0fe2 · 68a0fe2
1 parent cc2f567
commit 68a0fe2
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 5 deletions.
diff --git a/service/history/replication/task_processor.go b/service/history/replication/task_processor.go
@@ -255,11 +255,6 @@ func (p *taskProcessorImpl) cleanupAckedReplicationTasks() error {
 		metrics.ReplicationTasksLag,
 		time.Duration(p.shard.GetTransferMaxReadLevel()-minAckLevel),
 	)
-	// update replication queue ack level.
-	// this ack level currently use by Kafka replication
-	if err := p.shard.UpdateReplicatorAckLevel(minAckLevel); err != nil {
-		return err
-	}
 	return p.shard.GetExecutionManager().RangeCompleteReplicationTask(
 		&persistence.RangeCompleteReplicationTaskRequest{
 			InclusiveEndTaskID: minAckLevel,

diff --git a/service/history/replicatorQueueProcessor.go b/service/history/replicatorQueueProcessor.go
@@ -494,6 +494,12 @@ func (p *replicatorQueueProcessorImpl) getTasks(
 		p.logger.Error("error updating replication level for shard", tag.Error(err), tag.OperationFailed)
 	}
 
+	if err := p.shard.UpdateReplicatorAckLevel(
+		lastReadTaskID,
+	);  err != nil {
+		p.logger.Error("error updating kafka replication level", tag.Error(err), tag.OperationFailed)
+	}
+
 	return &replicator.ReplicationMessages{
 		ReplicationTasks:       replicationTasks,
 		HasMore:                common.BoolPtr(hasMore),