Repository: nats-io/nats-server
Pull request #1950 (2021-03-02 14:12:16)

update api types, remove meta peer by id
Signed-off-by: R.I.Pienaar <rip@devco.net>


/cc @nats-io/core



Patch:
diff --git a/server/jetstream.go b/server/jetstream.go
index 987aa216..5d476cc5 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -515,11 +515,6 @@ func (s *Server) migrateEphemerals() {
 			}
 		}
 	}
-
-	// Gove time for migration information to make it out of our server.
-	if len(consumers) > 0 {
-		time.Sleep(50 * time.Millisecond)
-	}
 }
 
 // Shutdown jetstream for this server.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 9b8f6289..68c33d40 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -448,19 +448,19 @@ type JSApiLeaderStepDownResponse struct {
 
 const JSApiLeaderStepDownResponseType = "io.nats.jetstream.api.v1.meta_leader_stepdown_response"
 
-// JSApiLeaderServerRemoveRequest will remove a peer from the system.
-type JSApiLeaderServerRemoveRequest struct {
-	// Server name of the peer to be removed.
+// JSApiMetaServerRemoveRequest will remove a peer from the meta group.
+type JSApiMetaServerRemoveRequest struct {
+	// Server ID of the peer to be removed.
 	Server string `json:"peer"`
 }
 
-// JSApiLeaderServerRemoveResponse is the response to a metaleader peer removal request.
-type JSApiLeaderServerRemoveResponse struct {
+// JSApiMetaServerRemoveResponse is the response to a peer removal request in the meta group.
+type JSApiMetaServerRemoveResponse struct {
 	ApiResponse
 	Success bool `json:"success,omitempty"`
 }
 
-const JSApiLeaderServerRemovalResponseType = "io.nats.jetstream.api.v1.meta_leader_server_removal"
+const JSApiMetaServerRemoveResponseType = "io.nats.jetstream.api.v1.meta_server_remove_response"
 
 // JSApiMsgGetRequest get a message request.
 type JSApiMsgGetRequest struct {
@@ -1799,7 +1799,7 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, subje
 		return
 	}
 
-	var resp = JSApiLeaderServerRemoveResponse{ApiResponse: ApiResponse{Type: JSApiLeaderServerRemovalResponseType}}
+	var resp = JSApiMetaServerRemoveResponse{ApiResponse: ApiResponse{Type: JSApiMetaServerRemoveResponseType}}
 
 	if isEmptyRequest(msg) {
 		resp.Error = jsBadRequestErr
@@ -1807,7 +1807,7 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, subje
 		return
 	}
 
-	var req JSApiLeaderServerRemoveRequest
+	var req JSApiMetaServerRemoveRequest
 	if err := json.Unmarshal(msg, &req); err != nil {
 		resp.Error = jsInvalidJSONErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1818,7 +1818,7 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, subje
 	js.mu.RLock()
 	for _, p := range cc.meta.Peers() {
 		si, ok := s.nodeToInfo.Load(p.ID)
-		if ok && si.(*nodeInfo).name == req.Server {
+		if ok && si.(*nodeInfo).id == req.Server {
 			found = p.ID
 			break
 		}


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 3dcb7282..34561dfc 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -2098,7 +2098,6 @@ func TestJetStreamClusterEphemeralConsumersNotReplicated(t *testing.T) {
 		t.Fatalf("Unexpected publish error: %v", err)
 	}
 	checkSubsPending(t, sub, 1)
-	sub.NextMsg(0)
 
 	if ci.Cluster == nil || len(ci.Cluster.Replicas) != 0 {
 		t.Fatalf("Expected ephemeral to be R=1, got %+v", ci.Cluster)
@@ -2116,13 +2115,13 @@ func TestJetStreamClusterEphemeralConsumersNotReplicated(t *testing.T) {
 	scl.Shutdown()
 	c.waitOnStreamLeader("$G", "foo")
 
+	// Let the consumer migrate and spin up.
+	time.Sleep(250 * time.Millisecond)
+
 	if _, err = js.Publish("foo", []byte("OK")); err != nil {
 		t.Fatalf("Unexpected publish error: %v", err)
 	}
-
-	if _, err := sub.NextMsg(500 * time.Millisecond); err != nil {
-		t.Logf("Expected to see another message, but behavior is optimistic so can fail")
-	}
+	checkSubsPending(t, sub, 2)
 }
 
 func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
@@ -2920,7 +2919,7 @@ func TestJetStreamClusterPeerRemovalAPI(t *testing.T) {
 	defer nc.Close()
 
 	// Expect error if unknown peer
-	req := &JSApiLeaderServerRemoveRequest{Server: "S-9"}
+	req := &JSApiMetaServerRemoveRequest{Server: "S-9"}
 	jsreq, err := json.Marshal(req)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -2929,7 +2928,7 @@ func TestJetStreamClusterPeerRemovalAPI(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	var resp JSApiLeaderServerRemoveResponse
+	var resp JSApiMetaServerRemoveResponse
 	if err := json.Unmarshal(rmsg.Data, &resp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2942,7 +2941,7 @@ func TestJetStreamClusterPeerRemovalAPI(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	req = &JSApiLeaderServerRemoveRequest{Server: "S-2"}
+	req = &JSApiMetaServerRemoveRequest{Server: c.serverByName("S-2").ID()}
 	jsreq, err = json.Marshal(req)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)