Fix producer panic by oldProducers (#598)

* Fix producer panic by oldProducers Signed-off-by: xiaolongran <[email protected]> * fix comments Signed-off-by: xiaolongran <[email protected]> * fix comments Signed-off-by: xiaolongran <[email protected]> * fix ci error Signed-off-by: xiaolongran <[email protected]>
apache · Aug 25, 2021 · baaf68d · baaf68d
1 parent bbffae2
commit baaf68d
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 6 deletions.
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
@@ -28,6 +28,7 @@ import (
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/pkg/errors"
 )
 
 const defaultNackRedeliveryDelay = 1 * time.Minute
@@ -266,16 +267,25 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 			return nil
 		}
 
+		if oldNumPartitions > newNumPartitions {
+			c.log.WithField("old_partitions", oldNumPartitions).
+				WithField("new_partitions", newNumPartitions).
+				Error("Does not support scaling down operations on topic partitions")
+			return errors.New("Does not support scaling down operations on topic partitions")
+		}
+
 		c.log.WithField("old_partitions", oldNumPartitions).
 			WithField("new_partitions", newNumPartitions).
 			Info("Changed number of partitions in topic")
 	}
 
 	c.consumers = make([]*partitionConsumer, newNumPartitions)
 
-	// Copy over the existing consumer instances
-	for i := 0; i < oldNumPartitions; i++ {
-		c.consumers[i] = oldConsumers[i]
+	if oldConsumers != nil {
+		// Copy over the existing consumer instances
+		for i := 0; i < oldNumPartitions; i++ {
+			c.consumers[i] = oldConsumers[i]
+		}
 	}
 
 	type ConsumerError struct {

diff --git a/pulsar/producer_impl.go b/pulsar/producer_impl.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -182,16 +183,26 @@ func (p *producer) internalCreatePartitionsProducers() error {
 			return nil
 		}
 
+		if oldNumPartitions > newNumPartitions {
+			p.log.WithField("old_partitions", oldNumPartitions).
+				WithField("new_partitions", newNumPartitions).
+				Error("Does not support scaling down operations on topic partitions")
+			return errors.New("Does not support scaling down operations on topic partitions")
+		}
+
 		p.log.WithField("old_partitions", oldNumPartitions).
 			WithField("new_partitions", newNumPartitions).
 			Info("Changed number of partitions in topic")
+
 	}
 
 	p.producers = make([]Producer, newNumPartitions)
 
-	// Copy over the existing consumer instances
-	for i := 0; i < oldNumPartitions; i++ {
-		p.producers[i] = oldProducers[i]
+	if oldProducers != nil {
+		// Copy over the existing consumer instances
+		for i := 0; i < oldNumPartitions; i++ {
+			p.producers[i] = oldProducers[i]
+		}
 	}
 
 	type ProducerError struct {