Merge eaa24ee into 93fa539

curator/fake_client.go

@@ -444,6 +444,15 @@ func (s *FakeZookeeper) GetApply(clientID FakeClientID) {
 
 // SetApply ...
 func (s *FakeZookeeper) SetApply(clientID FakeClientID) {
+	s.setApplyWithError(clientID, nil)
+}
+
+// SetApplyError ...
+func (s *FakeZookeeper) SetApplyError(clientID FakeClientID) {
+	s.setApplyWithError(clientID, zk.ErrConnectionClosed)
+}
+
+func (s *FakeZookeeper) setApplyWithError(clientID FakeClientID, err error) {
 	input := getActionWithType[SetInput](s, clientID, "Set")
 
 	node := s.findNode(input.Path)
@@ -465,6 +474,12 @@ func (s *FakeZookeeper) SetApply(clientID FakeClientID) {
 
 	s.notifyDataWatches(node, input.Path, zk.EventNodeDataChanged)
 
+	if err != nil {
+		input.Callback(zk.SetResponse{}, err)
+		s.ConnError(clientID)
+		return
+	}
+
 	input.Callback(zk.SetResponse{
 		Zxid: s.Zxid,
 		Stat: node.Stat,
@@ -473,6 +488,15 @@ func (s *FakeZookeeper) SetApply(clientID FakeClientID) {
 
 // DeleteApply ...
 func (s *FakeZookeeper) DeleteApply(clientID FakeClientID) {
+	s.deleteApplyWithError(clientID, nil)
+}
+
+// DeleteApplyError ...
+func (s *FakeZookeeper) DeleteApplyError(clientID FakeClientID) {
+	s.deleteApplyWithError(clientID, zk.ErrConnectionClosed)
+}
+
+func (s *FakeZookeeper) deleteApplyWithError(clientID FakeClientID, err error) {
 	input := getActionWithType[DeleteInput](s, clientID, "Delete")
 	node := s.findNode(input.Path)
 	if node == nil {
@@ -502,6 +526,12 @@ func (s *FakeZookeeper) DeleteApply(clientID FakeClientID) {
 	s.notifyChildrenWatches(parent, stdpath.Dir(input.Path))
 	s.notifyDataWatches(node, input.Path, zk.EventNodeDeleted)
 
+	if err != nil {
+		input.Callback(zk.DeleteResponse{}, err)
+		s.ConnError(clientID)
+		return
+	}
+
 	input.Callback(zk.DeleteResponse{
 		Zxid: s.Zxid,
 	}, nil)

curator/fake_client_test.go

@@ -1488,3 +1488,77 @@ func TestFakeClient_Create_Child_of_Ephemeral_Error(t *testing.T) {
 		zk.ErrNoChildrenForEphemerals,
 	}, errors)
 }
+
+func TestFakeClient__Set_Error(t *testing.T) {
+	c := newFakeClientTest()
+
+	var errors []error
+	callback := func(client Client) {
+		client.Create("/worker", []byte("data01"), zk.FlagEphemeral, func(resp zk.CreateResponse, err error) {
+			c.addStep("create-resp")
+			errors = append(errors, err)
+		})
+		client.Set("/worker", []byte("data02"), 0, func(resp zk.SetResponse, err error) {
+			c.addStep("set-resp")
+			errors = append(errors, err)
+		})
+	}
+
+	c.startCuratorClient1(func(sess *Session) {
+		sess.Run(callback)
+	})
+
+	c.store.Begin(client1)
+	c.store.CreateApply(client1)
+	c.store.SetApplyError(client1)
+
+	c.store.Retry(client1)
+
+	assert.Equal(t, []error{
+		nil,
+		zk.ErrConnectionClosed,
+	}, errors)
+	assert.Equal(t, []string{
+		"create-resp",
+		"set-resp",
+	}, c.steps)
+
+	node := c.store.Root.Children[0]
+	assert.Equal(t, "worker", node.Name)
+	assert.Equal(t, "data02", string(node.Data))
+}
+
+func TestFakeClient__Delete_Error(t *testing.T) {
+	c := newFakeClientTest()
+
+	var errors []error
+	callback := func(client Client) {
+		client.Create("/worker", []byte("data01"), zk.FlagEphemeral, func(resp zk.CreateResponse, err error) {
+			c.addStep("create-resp")
+			errors = append(errors, err)
+		})
+		client.Delete("/worker", 0, func(resp zk.DeleteResponse, err error) {
+			c.addStep("delete-resp")
+			errors = append(errors, err)
+		})
+	}
+
+	c.startCuratorClient1(func(sess *Session) {
+		sess.Run(callback)
+	})
+
+	c.store.Begin(client1)
+	c.store.CreateApply(client1)
+	c.store.DeleteApplyError(client1)
+
+	assert.Equal(t, []error{
+		nil,
+		zk.ErrConnectionClosed,
+	}, errors)
+	assert.Equal(t, []string{
+		"create-resp",
+		"delete-resp",
+	}, c.steps)
+
+	assert.Equal(t, 0, len(c.store.Root.Children))
+}

curator/fake_property.go

@@ -90,12 +90,53 @@ func (f *FakeZookeeperTester) doConnectionError(client FakeClientID) {
 	}
 }
 
+type runConfig struct {
+	opsErrorPercent float64
+}
+
+func (c runConfig) operationShouldError(randSource *rand.Rand) bool {
+	if c.opsErrorPercent == 0 {
+		return false
+	}
+	n := randSource.Intn(randMax)
+	end := int(c.opsErrorPercent / 100.0 * randMax)
+	if n < end {
+		return true
+	}
+	return false
+}
+
+func newRunConfig(options ...RunOption) runConfig {
+	conf := runConfig{
+		opsErrorPercent: 0.0,
+	}
+	for _, fn := range options {
+		fn(&conf)
+	}
+	return conf
+}
+
+// RunOption ...
+type RunOption func(conf *runConfig)
+
+// WithRunOperationErrorPercentage ...
+func WithRunOperationErrorPercentage(percent float64) RunOption {
+	return func(conf *runConfig) {
+		conf.opsErrorPercent = percent
+	}
+}
+
 // RunSessionExpiredAndConnectionError ...
+//
+//revive:disable-next-line:cognitive-complexity,cyclomatic
 func (f *FakeZookeeperTester) RunSessionExpiredAndConnectionError(
 	sessionExpiredPercentage float64,
 	connectionErrorPercentage float64,
 	numSteps int,
+	options ...RunOption,
 ) int {
+	conf := newRunConfig(options...)
+
 	sessionExpiredEnd := int(sessionExpiredPercentage / 100.0 * randMax)
 	connectionErrorEnd := int(connectionErrorPercentage / 100.0 * randMax)
 
@@ -123,15 +164,27 @@ func (f *FakeZookeeperTester) RunSessionExpiredAndConnectionError(
 		genericCmd := f.store.Pending[client][0]
 		switch genericCmd.(type) {
 		case CreateInput:
-			f.store.CreateApply(client)
+			if conf.operationShouldError(f.rand) {
+				f.store.CreateApplyError(client)
+			} else {
+				f.store.CreateApply(client)
+			}
 		case GetInput:
 			f.store.GetApply(client)
 		case ChildrenInput:
 			f.store.ChildrenApply(client)
 		case DeleteInput:
-			f.store.DeleteApply(client)
+			if conf.operationShouldError(f.rand) {
+				f.store.DeleteApplyError(client)
+			} else {
+				f.store.DeleteApply(client)
+			}
 		case SetInput:
-			f.store.SetApply(client)
+			if conf.operationShouldError(f.rand) {
+				f.store.SetApplyError(client)
+			} else {
+				f.store.SetApply(client)
+			}
 		case RetryInput:
 			f.store.Retry(client)
 		default:

curator/fake_property_test.go

@@ -3,6 +3,7 @@ package curator
 import (
 	"errors"
 	"fmt"
+	"math/rand"
 	"strconv"
 	"testing"
 	"time"
@@ -188,3 +189,43 @@ func TestFakeZookeeperTester_Master_Lock__Multi_Times(t *testing.T) {
 		)
 	}
 }
+
+func TestFakeZookeeperTester_Master_Lock__Multi_Times__With_Ops_Error(t *testing.T) {
+	for i := 0; i < 1000; i++ {
+		seed := time.Now().UnixNano()
+		fmt.Println("SEED:", seed)
+
+		store := NewFakeZookeeper()
+		tester := NewFakeZookeeperTester(store, []FakeClientID{
+			client1,
+			client2,
+			client3,
+		}, seed)
+
+		newSimpleLock(store, client1)
+		newSimpleLock(store, client2)
+		newSimpleLock(store, client3)
+
+		tester.Begin()
+
+		tester.RunSessionExpiredAndConnectionError(
+			10,
+			10,
+			1000,
+			WithRunOperationErrorPercentage(15),
+		)
+	}
+}
+
+func TestRunConfig(t *testing.T) {
+	c := newRunConfig(WithRunOperationErrorPercentage(20))
+
+	source := rand.New(rand.NewSource(1234))
+	trueCount := 0
+	for i := 0; i < 1000; i++ {
+		if c.operationShouldError(source) {
+			trueCount++
+		}
+	}
+	assert.Equal(t, 214, trueCount)
+}