Function Test_CDC_MultiNode

system_test/cdc_test.go:250–390

(t *testing.T)

}

func Test_CDC_MultiNode(t *testing.T) {
	testFn := func(t *testing.T, failRate int) {
		node1 := mustNewLeaderNode("node1")
		defer node1.Deprovision()
		node2 := mustNewNode("node2", false)
		defer node2.Deprovision()
		if err := node2.Join(node1); err != nil {
			t.Fatalf("node failed to join leader: %s", err.Error())
		}
		_, err := node2.WaitForLeader()
		if err != nil {
			t.Fatalf("failed waiting for leader: %s", err.Error())
		}
		node3 := mustNewNode("node3", false)
		defer node3.Deprovision()
		if err := node3.Join(node1); err != nil {
			t.Fatalf("node failed to join leader: %s", err.Error())
		}
		_, err = node3.WaitForLeader()
		if err != nil {
			t.Fatalf("failed waiting for leader: %s", err.Error())
		}

		testEndpoint := cdctest.NewHTTPTestServer()
		testEndpoint.SetFailRate(failRate)
		testEndpoint.Start()
		defer testEndpoint.Close()

		// Configure CDC service for each node.
		for _, node := range []*Node{node1, node2, node3} {
			cdcCluster := cdc.NewCDCCluster(node.Store, node.Cluster, node.Client)
			cdcService, err := cdc.NewService(node.ID, node.Dir, cdcCluster, mustCDCConfig(testEndpoint.URL()))
			if err != nil {
				panic(fmt.Sprintf("failed to create CDC service: %s", err.Error()))
			}
			node.CDC = cdcService
			node.CDC.Start()
			node.Store.EnableCDC(node.CDC.C(), nil, false)
		}

		node1.CDC.SetLeader(true)

		_, err = node1.Execute(`CREATE TABLE foo (id integer not null primary key, name text)`)
		if err != nil {
			t.Fatalf("failed to create table: %v", err)
		}
		_, err = node1.Execute(`INSERT INTO foo (id, name) VALUES (1, 'Alice')`)
		if err != nil {
			t.Fatalf("failed to insert data: %v", err)
		}
		_, err = node1.Execute(`UPDATE foo SET name = 'Alice Updated' WHERE id = 1`)
		if err != nil {
			t.Fatalf("failed to update data: %v", err)
		}
		_, err = node1.Execute(`DELETE FROM foo WHERE id = 1`)
		if err != nil {
			t.Fatalf("failed to delete data: %v", err)
		}