You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "LindaSummer (via GitHub)" <gi...@apache.org> on 2024/12/11 15:13:53 UTC
[PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
LindaSummer opened a new pull request, #2691:
URL: https://github.com/apache/kvrocks/pull/2691
# Issue
Close #2671
# Proposed Changes
Add integration test for losing master during SST file syncing.
Here is the steps.
1. Create master and slave instances.
2. Apply master and slave setnodeid.
3. Create a slow tcp proxy for master.
4. Setnodes to master and write data to master.
5. Setnodes to slave.
6. Wait for one second and close tcp proxy.
7. Try setnodes in slave and get duration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#discussion_r1881362778
##########
tests/gocase/integration/replication/replication_test.go:
##########
@@ -552,3 +552,60 @@ func TestFullSyncReplication(t *testing.T) {
require.Equal(t, "bar", slaveClient.Get(ctx, "foo").Val())
})
}
+
+func TestSlaveLostMaster(t *testing.T) {
+ // integration test for #2662 and #2671
+ ctx := context.Background()
+
+ masterSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "max-replication-mb": "1",
+ "rocksdb.compression": "no",
+ "rocksdb.write_buffer_size": "1",
+ "rocksdb.target_file_size_base": "1",
+ })
+ defer func() { masterSrv.Close() }()
+ masterClient := masterSrv.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+ masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+ replicaSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "replication-connect-timeout-ms": "5000",
+ "replication-recv-timeout-ms": "5100",
+ })
+ defer func() { replicaSrv.Close() }()
+ replicaClient := replicaSrv.NewClient()
+ // allow to run the read-only command in the replica
+ require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+ replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ newMasterPort := util.SimpleTCPProxy(proxyCtx, t, fmt.Sprintf("127.0.0.1:%d", masterSrv.Port()), true)
+
+ masterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, masterSrv.Port(), replicaNodeID, replicaSrv.Port(), masterNodeID)
+ clusterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+ unexistNodesInfo := fmt.Sprintf("%s 127.0.0.2 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", masterNodesInfo, "1").Err())
+ value := strings.Repeat("a", 128*1024)
+
+ for i := 0; i < 1024; i++ {
+ require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+ }
+
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodesInfo, "1").Err())
+
+ time.Sleep(2 * time.Second)
+ cancelProxy()
+ start := time.Now()
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", unexistNodesInfo, "2").Err())
+ duration := time.Since(start)
+ require.Less(t, duration, time.Second*6)
Review Comment:
Hi @mapleFU ,
Yes, I thought it should be in a [5, 6 or longer] range.
But I only got about one second in my own environment and make me confused.
Maybe we can set the upper bound to a conservative value just for integration test. 😊
Best Regards,
Edward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#issuecomment-2537696633
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2691) **Quality Gate passed**
Issues
 [5 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2691&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2691&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2691&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [57.4% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2691&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2691&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2691)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU merged PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#issuecomment-2537524415
Hi,
I'm not sure my case is correctly covered the scenario in related issue.
I only got about 1 second delay in my own machine.
Please let me know if there is any problem about the test case. 😊
Best Regards,
Edward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#discussion_r1881333140
##########
tests/gocase/integration/replication/replication_test.go:
##########
@@ -552,3 +552,60 @@ func TestFullSyncReplication(t *testing.T) {
require.Equal(t, "bar", slaveClient.Get(ctx, "foo").Val())
})
}
+
+func TestSlaveLostMaster(t *testing.T) {
+ // integration test for #2662 and #2671
+ ctx := context.Background()
+
+ masterSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "max-replication-mb": "1",
+ "rocksdb.compression": "no",
+ "rocksdb.write_buffer_size": "1",
+ "rocksdb.target_file_size_base": "1",
+ })
+ defer func() { masterSrv.Close() }()
+ masterClient := masterSrv.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+ masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+ replicaSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "replication-connect-timeout-ms": "5000",
+ "replication-recv-timeout-ms": "5100",
+ })
+ defer func() { replicaSrv.Close() }()
+ replicaClient := replicaSrv.NewClient()
+ // allow to run the read-only command in the replica
+ require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+ replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ newMasterPort := util.SimpleTCPProxy(proxyCtx, t, fmt.Sprintf("127.0.0.1:%d", masterSrv.Port()), true)
+
+ masterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, masterSrv.Port(), replicaNodeID, replicaSrv.Port(), masterNodeID)
+ clusterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+ unexistNodesInfo := fmt.Sprintf("%s 127.0.0.2 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", masterNodesInfo, "1").Err())
+ value := strings.Repeat("a", 128*1024)
+
+ for i := 0; i < 1024; i++ {
+ require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+ }
+
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodesInfo, "1").Err())
+
+ time.Sleep(2 * time.Second)
+ cancelProxy()
+ start := time.Now()
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", unexistNodesInfo, "2").Err())
+ duration := time.Since(start)
+ require.Less(t, duration, time.Second*6)
Review Comment:
Why the time out is 5000/5100 but the duration be 6 here 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#discussion_r1881362778
##########
tests/gocase/integration/replication/replication_test.go:
##########
@@ -552,3 +552,60 @@ func TestFullSyncReplication(t *testing.T) {
require.Equal(t, "bar", slaveClient.Get(ctx, "foo").Val())
})
}
+
+func TestSlaveLostMaster(t *testing.T) {
+ // integration test for #2662 and #2671
+ ctx := context.Background()
+
+ masterSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "max-replication-mb": "1",
+ "rocksdb.compression": "no",
+ "rocksdb.write_buffer_size": "1",
+ "rocksdb.target_file_size_base": "1",
+ })
+ defer func() { masterSrv.Close() }()
+ masterClient := masterSrv.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+ masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+ replicaSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "replication-connect-timeout-ms": "5000",
+ "replication-recv-timeout-ms": "5100",
+ })
+ defer func() { replicaSrv.Close() }()
+ replicaClient := replicaSrv.NewClient()
+ // allow to run the read-only command in the replica
+ require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+ replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ newMasterPort := util.SimpleTCPProxy(proxyCtx, t, fmt.Sprintf("127.0.0.1:%d", masterSrv.Port()), true)
+
+ masterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, masterSrv.Port(), replicaNodeID, replicaSrv.Port(), masterNodeID)
+ clusterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+ unexistNodesInfo := fmt.Sprintf("%s 127.0.0.2 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", masterNodesInfo, "1").Err())
+ value := strings.Repeat("a", 128*1024)
+
+ for i := 0; i < 1024; i++ {
+ require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+ }
+
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodesInfo, "1").Err())
+
+ time.Sleep(2 * time.Second)
+ cancelProxy()
+ start := time.Now()
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", unexistNodesInfo, "2").Err())
+ duration := time.Since(start)
+ require.Less(t, duration, time.Second*6)
Review Comment:
Hi @mapleFU ,
Yes, I thought it should be in a [5, 6 or longer] range.
But I only got about one second in my own environment and make me confused and not sure whether the case full covers the issue. 🤔
Maybe we can set the upper bound to a conservative value just for integration test. 😊
Best Regards,
Edward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#discussion_r1881347438
##########
tests/gocase/integration/replication/replication_test.go:
##########
@@ -552,3 +552,60 @@ func TestFullSyncReplication(t *testing.T) {
require.Equal(t, "bar", slaveClient.Get(ctx, "foo").Val())
})
}
+
+func TestSlaveLostMaster(t *testing.T) {
+ // integration test for #2662 and #2671
+ ctx := context.Background()
+
+ masterSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "max-replication-mb": "1",
+ "rocksdb.compression": "no",
+ "rocksdb.write_buffer_size": "1",
+ "rocksdb.target_file_size_base": "1",
+ })
+ defer func() { masterSrv.Close() }()
+ masterClient := masterSrv.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+ masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+ replicaSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "replication-connect-timeout-ms": "5000",
+ "replication-recv-timeout-ms": "5100",
+ })
+ defer func() { replicaSrv.Close() }()
+ replicaClient := replicaSrv.NewClient()
+ // allow to run the read-only command in the replica
+ require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+ replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ newMasterPort := util.SimpleTCPProxy(proxyCtx, t, fmt.Sprintf("127.0.0.1:%d", masterSrv.Port()), true)
+
+ masterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, masterSrv.Port(), replicaNodeID, replicaSrv.Port(), masterNodeID)
+ clusterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+ unexistNodesInfo := fmt.Sprintf("%s 127.0.0.2 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", masterNodesInfo, "1").Err())
+ value := strings.Repeat("a", 128*1024)
+
+ for i := 0; i < 1024; i++ {
+ require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+ }
+
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodesInfo, "1").Err())
+
+ time.Sleep(2 * time.Second)
+ cancelProxy()
+ start := time.Now()
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", unexistNodesInfo, "2").Err())
+ duration := time.Since(start)
+ require.Less(t, duration, time.Second*6)
Review Comment:
What would a duration really looks like? Would it:
1. Force greater than 5?
2. Always less than 6? (I guess this is not guranteed ...)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#discussion_r1881342886
##########
tests/gocase/integration/replication/replication_test.go:
##########
@@ -552,3 +552,60 @@ func TestFullSyncReplication(t *testing.T) {
require.Equal(t, "bar", slaveClient.Get(ctx, "foo").Val())
})
}
+
+func TestSlaveLostMaster(t *testing.T) {
+ // integration test for #2662 and #2671
+ ctx := context.Background()
+
+ masterSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "max-replication-mb": "1",
+ "rocksdb.compression": "no",
+ "rocksdb.write_buffer_size": "1",
+ "rocksdb.target_file_size_base": "1",
+ })
+ defer func() { masterSrv.Close() }()
+ masterClient := masterSrv.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+ masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+ replicaSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "replication-connect-timeout-ms": "5000",
+ "replication-recv-timeout-ms": "5100",
+ })
+ defer func() { replicaSrv.Close() }()
+ replicaClient := replicaSrv.NewClient()
+ // allow to run the read-only command in the replica
+ require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+ replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ newMasterPort := util.SimpleTCPProxy(proxyCtx, t, fmt.Sprintf("127.0.0.1:%d", masterSrv.Port()), true)
+
+ masterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, masterSrv.Port(), replicaNodeID, replicaSrv.Port(), masterNodeID)
+ clusterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+ unexistNodesInfo := fmt.Sprintf("%s 127.0.0.2 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", masterNodesInfo, "1").Err())
+ value := strings.Repeat("a", 128*1024)
+
+ for i := 0; i < 1024; i++ {
+ require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+ }
+
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodesInfo, "1").Err())
+
+ time.Sleep(2 * time.Second)
+ cancelProxy()
+ start := time.Now()
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", unexistNodesInfo, "2").Err())
+ duration := time.Since(start)
+ require.Less(t, duration, time.Second*6)
Review Comment:
Hi @mapleFU ,
The time duration may not be a precise value due to network.
So I reserved 1 second to reduce flaky failure. 😊
Best Regards,
Edward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#discussion_r1881362778
##########
tests/gocase/integration/replication/replication_test.go:
##########
@@ -552,3 +552,60 @@ func TestFullSyncReplication(t *testing.T) {
require.Equal(t, "bar", slaveClient.Get(ctx, "foo").Val())
})
}
+
+func TestSlaveLostMaster(t *testing.T) {
+ // integration test for #2662 and #2671
+ ctx := context.Background()
+
+ masterSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "max-replication-mb": "1",
+ "rocksdb.compression": "no",
+ "rocksdb.write_buffer_size": "1",
+ "rocksdb.target_file_size_base": "1",
+ })
+ defer func() { masterSrv.Close() }()
+ masterClient := masterSrv.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+ masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+ replicaSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "replication-connect-timeout-ms": "5000",
+ "replication-recv-timeout-ms": "5100",
+ })
+ defer func() { replicaSrv.Close() }()
+ replicaClient := replicaSrv.NewClient()
+ // allow to run the read-only command in the replica
+ require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+ replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ newMasterPort := util.SimpleTCPProxy(proxyCtx, t, fmt.Sprintf("127.0.0.1:%d", masterSrv.Port()), true)
+
+ masterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, masterSrv.Port(), replicaNodeID, replicaSrv.Port(), masterNodeID)
+ clusterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+ unexistNodesInfo := fmt.Sprintf("%s 127.0.0.2 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", masterNodesInfo, "1").Err())
+ value := strings.Repeat("a", 128*1024)
+
+ for i := 0; i < 1024; i++ {
+ require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+ }
+
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodesInfo, "1").Err())
+
+ time.Sleep(2 * time.Second)
+ cancelProxy()
+ start := time.Now()
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", unexistNodesInfo, "2").Err())
+ duration := time.Since(start)
+ require.Less(t, duration, time.Second*6)
Review Comment:
Hi @mapleFU ,
Yes, I thought it should be in a [5, 6 or longer] range.
But I only got about one second in my own environment and make me confused.
Maybe we can the upper bound to a conservative value just for integration test.
Best Regards,
Edward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(integration): add integration test for master lost during syncing sst files. [kvrocks]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #2691:
URL: https://github.com/apache/kvrocks/pull/2691#discussion_r1881466925
##########
tests/gocase/integration/replication/replication_test.go:
##########
@@ -552,3 +552,60 @@ func TestFullSyncReplication(t *testing.T) {
require.Equal(t, "bar", slaveClient.Get(ctx, "foo").Val())
})
}
+
+func TestSlaveLostMaster(t *testing.T) {
+ // integration test for #2662 and #2671
+ ctx := context.Background()
+
+ masterSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "max-replication-mb": "1",
+ "rocksdb.compression": "no",
+ "rocksdb.write_buffer_size": "1",
+ "rocksdb.target_file_size_base": "1",
+ })
+ defer func() { masterSrv.Close() }()
+ masterClient := masterSrv.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+ masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+ replicaSrv := util.StartServer(t, map[string]string{
+ "cluster-enabled": "yes",
+ "replication-connect-timeout-ms": "5000",
+ "replication-recv-timeout-ms": "5100",
+ })
+ defer func() { replicaSrv.Close() }()
+ replicaClient := replicaSrv.NewClient()
+ // allow to run the read-only command in the replica
+ require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+ replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ newMasterPort := util.SimpleTCPProxy(proxyCtx, t, fmt.Sprintf("127.0.0.1:%d", masterSrv.Port()), true)
+
+ masterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, masterSrv.Port(), replicaNodeID, replicaSrv.Port(), masterNodeID)
+ clusterNodesInfo := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+ unexistNodesInfo := fmt.Sprintf("%s 127.0.0.2 %d master - 0-16383\n%s 127.0.0.1 %d slave %s",
+ masterNodeID, newMasterPort, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+ require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", masterNodesInfo, "1").Err())
+ value := strings.Repeat("a", 128*1024)
+
+ for i := 0; i < 1024; i++ {
+ require.NoError(t, masterClient.Set(ctx, fmt.Sprintf("key%d", i), value, 0).Err())
+ }
+
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodesInfo, "1").Err())
+
+ time.Sleep(2 * time.Second)
+ cancelProxy()
+ start := time.Now()
+ require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", unexistNodesInfo, "2").Err())
+ duration := time.Since(start)
+ require.Less(t, duration, time.Second*6)
Review Comment:
> But I only got about one second in my own environment and make me confused and not sure whether the case full covers the issue. 🤔
Thanks! I'll dive in my env later, we can first move forward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org