[azeventhubs] Fixing broken behavior in Processor (Azure#22833)

Explicitly disallow some sharp edges with Run():
  * If Run() is currently active for this instance
  * If Run() has stopped for this instance.
  * Now we close the channel so NextPartitionClient() also returns nil if the Processor has been stopped.
* Fixing the credential test - looks like a change in validation in the credential itself exposed some silliness on my end.

Fixes Azure#22785

Author: richardpark-msft

sdk/messaging/azeventhubs/CHANGELOG.md

@@ -1,14 +1,14 @@
 # Release History
 
-## 1.1.1 (Unreleased)
-
-### Features Added
-
-### Breaking Changes
+## 1.2.0 (2024-05-07)
 
 ### Bugs Fixed
 
-### Other Changes
+Processor.Run had unclear behavior for some cases:
+- Run() now returns an explicit error when called more than once on a single 
+  Processor instance or if multiple Run calls are made concurrently. (PR#22833)
+- NextProcessorClient now properly terminates (and returns nil) if called on a 
+  stopped Processor. (PR#22833)
 
 ## 1.1.0 (2024-04-02)
 

sdk/messaging/azeventhubs/example_enabling_logging_test.go

@@ -23,10 +23,6 @@ func Example_enableLogging() {
 	)
 
 	fmt.Printf("Logging enabled\n")
-
-	// Output:
-	// Logging enabled
-	//
 }
 
 func printLoggedEvent(event azlog.Event, s string) {

sdk/messaging/azeventhubs/internal/constants.go

@@ -4,4 +4,4 @@
 package internal
 
 // Version is the semantic version number
-const Version = "v1.1.1"
+const Version = "v1.2.0"

sdk/messaging/azeventhubs/internal/test/test_helpers.go

@@ -136,8 +136,6 @@ func GetConnectionParamsForTest(t *testing.T) ConnectionParamsForTest {
 		ResourceGroup:              envVars["RESOURCE_GROUP"],
 		StorageConnectionString:    envVars["CHECKPOINTSTORE_STORAGE_CONNECTION_STRING"],
 		SubscriptionID:             envVars["AZURE_SUBSCRIPTION_ID"],
-		TenantID:                   envVars["AZURE_TENANT_ID"],
-		ClientID:                   envVars["AZURE_CLIENT_ID"],
 	}
 }
 

sdk/messaging/azeventhubs/processor.go

@@ -5,6 +5,7 @@ package azeventhubs
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/rand"
 	"sync"
@@ -78,6 +79,14 @@ type StartPositions struct {
 	Default StartPosition
 }
 
+type state int32
+
+const (
+	stateNone    state = 0
+	stateStopped state = 1
+	stateRunning state = 2
+)
+
 // Processor uses a [ConsumerClient] and [CheckpointStore] to provide automatic
 // load balancing between multiple Processor instances, even in separate
 // processes or on separate machines.
@@ -87,6 +96,9 @@ type StartPositions struct {
 //
 // [example_consuming_with_checkpoints_test.go]: https://github.com/Azure/azure-sdk-for-go/blob/main/sdk/messaging/azeventhubs/example_consuming_with_checkpoints_test.go
 type Processor struct {
+	stateMu sync.Mutex
+	state   state
+
 	ownershipUpdateInterval time.Duration
 	defaultStartPositions   StartPositions
 	checkpointStore         CheckpointStore
@@ -97,10 +109,10 @@ type Processor struct {
 	consumerClient consumerClientForProcessor
 
 	nextClients           chan *ProcessorPartitionClient
+	nextClientsReady      chan struct{}
 	consumerClientDetails consumerClientDetails
 
-	runCalled chan struct{}
-	lb        *processorLoadBalancer
+	lb *processorLoadBalancer
 
 	// claimedOwnerships is set to whatever our current ownerships are. The underlying
 	// value is a []Ownership.
@@ -173,12 +185,13 @@ func newProcessorImpl(consumerClient consumerClientForProcessor, checkpointStore
 		},
 		prefetch:              options.Prefetch,
 		consumerClientDetails: consumerClient.getDetails(),
-		runCalled:             make(chan struct{}),
+		nextClientsReady:      make(chan struct{}),
 		lb:                    newProcessorLoadBalancer(checkpointStore, consumerClient.getDetails(), strategy, partitionDurationExpiration),
 		currentOwnerships:     currentOwnerships,
 
-		// `nextClients` will be initialized when the user calls Run() since it needs to query the #
-		// of partitions on the Event Hub.
+		// `nextClients` will be properly initialized when the user calls
+		// Run() since it needs to query the # of partitions on the Event Hub.
+		nextClients: make(chan *ProcessorPartitionClient),
 	}, nil
 }
 
@@ -196,7 +209,7 @@ func (p *Processor) NextPartitionClient(ctx context.Context) *ProcessorPartition
 	select {
 	case <-ctx.Done():
 		return nil
-	case <-p.runCalled:
+	case <-p.nextClientsReady:
 	}
 
 	select {
@@ -207,12 +220,30 @@ func (p *Processor) NextPartitionClient(ctx context.Context) *ProcessorPartition
 	}
 }
 
+func (p *Processor) checkState() error {
+	switch p.state {
+	case stateNone:
+		// not running so we can start. And lock out any other users.
+		p.state = stateRunning
+		return nil
+	case stateRunning:
+		return errors.New("the Processor is currently running. Concurrent calls to Run() are not allowed.")
+	case stateStopped:
+		return errors.New("the Processor has been stopped. Create a new instance to start processing again")
+	default:
+		return fmt.Errorf("unhandled state value %v", p.state)
+	}
+}
+
 // Run handles the load balancing loop, blocking until the passed in context is cancelled
 // or it encounters an unrecoverable error. On cancellation, it will return a nil error.
 //
 // This function should run for the lifetime of your application, or for as long as you want
 // to continue to claim and process partitions.
 //
+// Once a Processor has been stopped it cannot be restarted and a new instance must
+// be created.
+//
 // As partitions are claimed new [ProcessorPartitionClient] instances will be returned from
 // [Processor.NextPartitionClient]. This can happen at any time, based on new Processor instances
 // coming online, as well as other Processors exiting.
@@ -225,7 +256,15 @@ func (p *Processor) NextPartitionClient(ctx context.Context) *ProcessorPartition
 //
 // [example_consuming_with_checkpoints_test.go]: https://github.com/Azure/azure-sdk-for-go/blob/main/sdk/messaging/azeventhubs/example_consuming_with_checkpoints_test.go
 func (p *Processor) Run(ctx context.Context) error {
-	err := p.runImpl(ctx)
+	p.stateMu.Lock()
+	err := p.checkState()
+	p.stateMu.Unlock()
+
+	if err != nil {
+		return err
+	}
+
+	err = p.runImpl(ctx)
 
 	// the context is the proper way to close down the Run() loop, so it's not
 	// an error and doesn't need to be returned.
@@ -241,7 +280,7 @@ func (p *Processor) runImpl(ctx context.Context) error {
 	defer func() {
 		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
 		defer cancel()
-		p.closeConsumers(ctx, consumers)
+		p.close(ctx, consumers)
 	}()
 
 	// size the channel to the # of partitions. We can never exceed this size since
@@ -287,7 +326,7 @@ func (p *Processor) initNextClientsCh(ctx context.Context) (EventHubProperties,
 	}
 
 	p.nextClients = make(chan *ProcessorPartitionClient, len(eventHubProperties.PartitionIDs))
-	close(p.runCalled)
+	close(p.nextClientsReady)
 
 	return eventHubProperties, nil
 }
@@ -433,7 +472,7 @@ func (p *Processor) getCheckpointsMap(ctx context.Context) (map[string]Checkpoin
 	return m, nil
 }
 
-func (p *Processor) closeConsumers(ctx context.Context, consumersMap *sync.Map) {
+func (p *Processor) close(ctx context.Context, consumersMap *sync.Map) {
 	consumersMap.Range(func(key, value any) bool {
 		client := value.(*ProcessorPartitionClient)
 
@@ -455,6 +494,20 @@ func (p *Processor) closeConsumers(ctx context.Context, consumersMap *sync.Map)
 	if err != nil {
 		azlog.Writef(EventConsumer, "Failed to relinquish ownerships. New processors will have to wait for ownerships to expire: %s", err.Error())
 	}
+
+	p.stateMu.Lock()
+	p.state = stateStopped
+	p.stateMu.Unlock()
+
+	// NextPartitionClient() will quit out now that p.nextClients is closed.
+	close(p.nextClients)
+
+	select {
+	case <-p.nextClientsReady:
+		// already closed
+	default:
+		close(p.nextClientsReady)
+	}
 }
 
 // relinquishedOwnershipID indicates that a partition is immediately available, similar to