fix: add service not ready check

apache · Apr 11, 2022 · 545f579 · 545f579
1 parent ae400ca
commit 545f579
Showing 1 changed file with 13 additions and 0 deletions.
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
@@ -522,10 +522,13 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 		c.handleResponse(cmd.ProducerSuccess.GetRequestId(), cmd)
 
 	case pb.BaseCommand_PARTITIONED_METADATA_RESPONSE:
+		partitionMetadataResponse := cmd.PartitionMetadataResponse
+		c.checkServerError(partitionMetadataResponse.Error)
 		c.handleResponse(cmd.PartitionMetadataResponse.GetRequestId(), cmd)
 
 	case pb.BaseCommand_LOOKUP_RESPONSE:
 		lookupResult := cmd.LookupTopicResponse
+		c.checkServerError(lookupResult.Error)
 		c.handleResponse(lookupResult.GetRequestId(), cmd)
 
 	case pb.BaseCommand_CONSUMER_STATS_RESPONSE:
@@ -574,6 +577,16 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 	}
 }
 
+func (c *connection) checkServerError(err *pb.ServerError) {
+	if err == nil {
+		return
+	}
+
+	if *err == pb.ServerError_ServiceNotReady {
+		c.Close()
+	}
+}
+
 func (c *connection) Write(data Buffer) {
 	c.writeRequestsCh <- data
 }