Handle more edge case when apply events #836

wxing1292 · 2018-06-11T20:15:38Z

UT for history replicator
Remove deadcode
Bugfix start & signal workflow

samarabbas · 2018-06-11T21:54:57Z

service/history/historyReplicator.go

@@ -37,6 +38,9 @@ import (
 	"github.com/uber/cadence/common/persistence"
 )

+var errRetry = &shared.RetryTaskError{Message: "History replicator retry error"}
+var errDLQ = errors.New("History replicator DQL error")


nit: Fix "DQL" on error message.

samarabbas · 2018-06-11T21:57:41Z

service/history/historyReplicator.go

@@ -37,6 +38,9 @@ import (
 	"github.com/uber/cadence/common/persistence"
 )

+var errRetry = &shared.RetryTaskError{Message: "History replicator retry error"}


Can you define all errors within the same var section. There is already one error for RetryTaskError.

samarabbas · 2018-06-11T22:01:57Z

service/history/historyReplicator.go

@@ -77,32 +81,39 @@ func newHistoryReplicator(shard ShardContext, historyEngine *historyEngineImpl,
 }

 func (r *historyReplicator) ApplyEvents(request *h.ReplicateEventsRequest) (retError error) {
+	defer func() {
+		if _, ok := retError.(*shared.EntityNotExistsError); ok {
+			retError = errRetry


This catch all error could make debugging little harder. I think we should also include the actual message here. Or atleast log a warning before returning the catch all error.

samarabbas · 2018-06-11T22:13:07Z

service/history/historyReplicator.go

-
-		// GetWorkflowExecution failed with some transient error.  Return err so we can retry the task later
-		if _, ok := err.(*shared.EntityNotExistsError); !ok {
+		} else if _, ok := err.(*shared.EntityNotExistsError); !ok {


why converting to else-if?

samarabbas · 2018-06-11T22:16:18Z

service/history/historyReplicator.go

-			if _, ok := err.(*shared.EntityNotExistsError); ok {
-				return ErrRetryEntityNotExists
+			if _, ok := err.(*shared.EntityNotExistsError); !ok {
+				return err


ErrRetryEntityNotExists error may not be needed if you remove this.

samarabbas · 2018-06-11T22:19:22Z

service/history/historyReplicator.go

+	msBuilder.executionInfo.LastFirstEventID = firstEvent.GetEventId()
+	msBuilder.executionInfo.NextEventID = lastEvent.GetEventId() + 1
+	incomingVersion := firstEvent.GetVersion()
+	replicationState := &persistence.ReplicationState{


Probably we should get rid of this and just call:
msBuilder.updateReplicationStateLastEventID

samarabbas · 2018-06-12T17:00:58Z

service/history/historyReplicator.go

+	executionInfo.LastFirstEventID = firstEvent.GetEventId()
+	executionInfo.NextEventID = lastEvent.GetEventId() + 1
+	incomingVersion := firstEvent.GetVersion()
+	msBuilder.UpdateReplicationStateLastEventID("", incomingVersion, lastEvent.GetEventId())


I think we should use the sourceCluster from replication task instead of empty string.

samarabbas · 2018-06-12T17:10:16Z

service/history/historyReplicator.go

-			_, err = createWorkflow(isBrandNew, errExist.RunID)
-		}
-
+		requestID := uuid.New()


Can we move all this logic into ApplyStartEvent? We can simplify this logic to deal with only other events and have an assertion that it should not be called for WorkflowExecutionStartedEvent.

actually, i think we should keep this logic like this.

all other functions, other than ApplyReplicationTask should prepare the mutable state.
ApplyReplicationTask should apply events to mutable state. and accordingly, create workflow or update workflow.
this can be useful if say, we want to apply events from a remote cluster to recover from a disaster, or simple reply a workflow using existing events for debugging reason

samarabbas · 2018-06-12T17:39:20Z

service/history/historyReplicator.go

+
+	// we can also use the start version
+	if currentLastWriteVersion > incomingVersion {
+		logger.Infof("Dropping replication task. Current RunID: %v, Current LastWriteVersion: %v, Incoming Version: %v.",


increment the 'StaleReplicationEventsCounter' counter.

added all 3 types:
HistoryConflictsCounter
StaleReplicationEventsCounter
BufferedReplicationTaskCounter

samarabbas · 2018-06-12T17:50:12Z

service/history/historyReplicator.go

+	if rState.LastWriteVersion > incomingVersion {
+		// Replication state is already on a higher version, we can drop this event
+		// TODO: We need to replay external events like signal to the new version
+		logger.Warnf("Dropping stale replication task. CurrentV: %v, LastWriteV: %v, LastWriteEvent: %v, IncomingV: %v.",


increment the 'StaleReplicationEventsCounter' counter.

samarabbas · 2018-06-12T17:52:10Z

service/history/historyReplicator.go

+		logger.Warnf("Dropping stale replication task. CurrentV: %v, LastWriteV: %v, LastWriteEvent: %v, IncomingV: %v.",
+			rState.CurrentVersion, rState.LastWriteVersion, rState.LastWriteEventID, incomingVersion)
+		return nil, nil
+	} else if rState.LastWriteVersion < incomingVersion {


Can we change 'else if' to 'if'

samarabbas · 2018-06-12T17:53:20Z

service/history/historyReplicator.go

+			logger.Errorf("No ReplicationInfo Found For Previous Active Cluster. Previous Active Cluster: %v, Request Source Cluster: %v, Request ReplicationInfo: %v.",
+				previousActiveCluster, request.GetSourceCluster(), request.ReplicationInfo)
+			// TODO: Handle missing replication information, #840
+			return nil, ErrMissingReplicationInfo


Can you also include the existing comment for this part of the code.

copy pasted the existing one: "// Returning BadRequestError to force the message to land into DLQ"
is this the comment you are talking about?

samarabbas · 2018-06-12T17:54:15Z

service/history/historyReplicator.go

-			r.metricsClient.IncCounter(metrics.ReplicateHistoryEventsScope, metrics.BufferedReplicationTaskCounter)
+		// Detect conflict
+		if ri.GetLastEventId() < rState.LastWriteEventID {
+			logger.Infof("Conflict detected.  State: {CurrentV: %v, LastWriteV: %v, LastWriteEvent: %v}, ReplicationInfo: {PrevActiveCluster: %v, V: %v, LastEventID: %v}",


increment 'HistoryConflictsCounter' counter.

samarabbas · 2018-06-12T17:55:28Z

service/history/historyReplicator.go

-				msBuilder.GetNextEventID(), firstEvent.GetEventId())
-			r.metricsClient.IncCounter(metrics.ReplicateHistoryEventsScope, metrics.BufferedReplicationTaskCounter)
+		// Detect conflict
+		if ri.GetLastEventId() < rState.LastWriteEventID {


you changed '!=' to '<'. Whats the reasoning? Can we capture that as a comment.

samarabbas · 2018-06-12T17:57:47Z

service/history/historyReplicator.go

 			}
+		} else if ri.GetLastEventId() > rState.LastWriteEventID {
+			// TODO handle this case
+			logger.Errorf("Conflict detected.  State: {CurrentV: %v, LastWriteV: %v, LastWriteEvent: %v}, ReplicationInfo: {PrevActiveCluster: %v, V: %v, LastEventID: %v}",


I don't think I get this. Can you explain?

say if we have a bug, this case can happen, we should return some unique error and do logging

samarabbas · 2018-06-12T18:20:38Z

service/history/historyReplicator.go

+			logger.Errorf("Conflict detected.  State: {CurrentV: %v, LastWriteV: %v, LastWriteEvent: %v}, ReplicationInfo: {PrevActiveCluster: %v, V: %v, LastEventID: %v}",
+				rState.CurrentVersion, rState.LastWriteVersion, rState.LastWriteEventID,
+				previousActiveCluster, ri.GetVersion(), ri.GetLastEventId())
+			return nil, ErrMissingReplicationInfo


Let's use a different error here.

done
// ErrCorruptedReplicationInfo is returned when replication task has corrupted replication information from source cluster
ErrCorruptedReplicationInfo = &shared.BadRequestError{Message: "replication task is has corrupted cluster replication info"}

samarabbas · 2018-06-12T18:21:11Z

service/history/historyReplicator.go

+	firstEventID := request.GetFirstEventId()
+	if firstEventID < msBuilder.GetNextEventID() {
+		// duplicate replication task
+		replicationState := msBuilder.GetReplicationState()


increment 'StaleReplicationEventsCounter' metric counter

samarabbas · 2018-06-12T18:22:05Z

service/history/historyReplicator.go

+		logger.Debugf("Dropping replication task.  State: {NextEvent: %v, Version: %v, LastWriteV: %v, LastWriteEvent: %v}",
+			msBuilder.GetNextEventID(), replicationState.CurrentVersion, replicationState.LastWriteVersion, replicationState.LastWriteEventID)
+		return nil
+	} else if firstEventID > msBuilder.GetNextEventID() {


change 'else if' to 'if'

samarabbas · 2018-06-12T18:23:14Z

service/history/historyReplicator.go

+		logger.Debugf("Buffer out of order replication task.  NextEvent: %v, FirstEvent: %v",
+			msBuilder.GetNextEventID(), firstEventID)
+
+		err = msBuilder.BufferReplicationTask(request)


increment 'BufferedReplicationTaskCounter' metric counter

samarabbas

Mostly looks good. Can you address my minor comments? It is good to land once you address those.

wxing1292 requested a review from samarabbas June 11, 2018 20:15

wxing1292 force-pushed the hReplictor branch 2 times, most recently from e236d8f to e19bc84 Compare June 11, 2018 21:42

Handle more edge case when apply events

efa021d

wxing1292 force-pushed the hReplictor branch from e19bc84 to efa021d Compare June 11, 2018 21:59

samarabbas reviewed Jun 11, 2018

View reviewed changes

Wenquan Xing added 2 commits June 12, 2018 01:58

adding UT

17873e4

Merge branch 'master' into hReplictor

4037424

samarabbas reviewed Jun 12, 2018

View reviewed changes

Wenquan Xing added 2 commits June 12, 2018 11:35

address comments

68e35ff

address comments

b5c3b29

samarabbas approved these changes Jun 12, 2018

View reviewed changes

wxing1292 merged commit 583fe1c into master Jun 12, 2018

wxing1292 deleted the hReplictor branch June 12, 2018 19:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Handle more edge case when apply events #836

Handle more edge case when apply events #836

wxing1292 commented Jun 11, 2018 •

edited

Loading

samarabbas Jun 11, 2018

samarabbas Jun 11, 2018

samarabbas Jun 11, 2018

samarabbas Jun 11, 2018

samarabbas Jun 11, 2018

samarabbas Jun 11, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018 •

edited

Loading

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas Jun 12, 2018 •

edited

Loading

samarabbas Jun 12, 2018

wxing1292 Jun 12, 2018

samarabbas left a comment

Handle more edge case when apply events #836

Handle more edge case when apply events #836

Conversation

wxing1292 commented Jun 11, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samarabbas Jun 12, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samarabbas Jun 12, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samarabbas left a comment

Choose a reason for hiding this comment

wxing1292 commented Jun 11, 2018 •

edited

Loading

samarabbas Jun 12, 2018 •

edited

Loading

samarabbas Jun 12, 2018 •

edited

Loading