Eventhub Receiver issue about azure-sdk-for-go HOT 6 CLOSED

Thank you for your feedback. Tagging and routing to the team member best able to assist.

from azure-sdk-for-go.

Here is the dummy test code which i have implemented here

`func main() {

eventHubConnectionString := 
eventHubName :=

storageConnectionString :=
storageContainerName :=

// create a container client using a connection string and container name
checkClient, _ := container.NewClientFromConnectionString(storageConnectionString, storageContainerName, nil)

// create a checkpoint store that will be used by the event hub
checkpointStore, err := checkpoints.NewBlobStore(checkClient, nil)

if err != nil {
	panic(err)
}

// create a consumer client using a connection string to the namespace and the event hub
consumerClient, err := azeventhubs.NewConsumerClientFromConnectionString(eventHubConnectionString, eventHubName, azeventhubs.DefaultConsumerGroup, nil)

if err != nil {
	panic(err)
}

defer consumerClient.Close(context.TODO())

// create a processor to receive and process events
processor, err := azeventhubs.NewProcessor(consumerClient, checkpointStore, nil)

if err != nil {
	panic(err)
}

//  for each partition in the event hub, create a partition client with processEvents as the function to process events
dispatchPartitionClients := func() {
	for {
		partitionClient := processor.NextPartitionClient(context.TODO())

		if partitionClient == nil {
			break
		}

		go func() {
			if err := processEvents(partitionClient); err != nil {
				panic(err)
			}
		}()
	}
}

// run all partition clients
go dispatchPartitionClients()

processorCtx, processorCancel := context.WithCancel(context.TODO())
defer processorCancel()

if err := processor.Run(processorCtx); err != nil {
	panic(err)
}

}

func processEvents(partitionClient *azeventhubs.ProcessorPartitionClient) error {
defer closePartitionResources(partitionClient)
for {
receiveCtx, receiveCtxCancel := context.WithTimeout(context.Background(), time.Minute)
events, err := partitionClient.ReceiveEvents(receiveCtx, 100, nil)
receiveCtxCancel()

	if err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}

	fmt.Printf("Processing %d event(s)\n", len(events))

	for _, event := range events {
		fmt.Printf("Event received with body %v\n", string(event.Body))
	}
	if len(events) != 0 {
		if err := partitionClient.UpdateCheckpoint(context.Background(), events[len(events)-1], nil); err != nil {
			return err
		}
	}
}

}

func closePartitionResources(partitionClient *azeventhubs.ProcessorPartitionClient) {
defer partitionClient.Close(context.TODO())
}`

please let me know if any partitioning issue or what

from azure-sdk-for-go.

like incoming messsage is coming 220 and my receiver is receiving 7.20 k data

The balance there seems quite off!

Let's start off with some questions, see if we can eliminate some causes:

1. Are you seeing duplicate messages?
2. Are you seeing any failures from partitionClient.UpdateCheckpoint? If that function fails restarts will end up consuming the same events again.
3. Are you running multiple Processor instances? And if so, are all of those instances using the same consumer group?
4. Are you using the same Azure Storage Blob container for each run of the Processor?

from azure-sdk-for-go.

Hi Team I am facing a similar issue where Incoming message count > Outgoing message count (Seems like these message just get dropped. I tried multiple runs.):

Answering @richardpark-msft questions above.

1. No
2. No
3. Yes multiple instances using the same consumer group. I run 5 tasks which have consumers consuming messages from my eventhub's 32 partitions (Each consumer is assigned 6-7 partitions).
4. Yes

from azure-sdk-for-go.

Issue got fixed, i am closing this issue .
Thanks for your support.

from azure-sdk-for-go.

Issue got fixed, i am closing this issue . Thanks for your support.

@devak8, I feel like I dropped the ball here but somehow it worked out. I'm curious - did you find something on your end that seemed like the culprit or did the problem just go away?

One thing I was curious about (when thinking about this more) is if you were having some bad interaction with the prefetch cache. I need to write about it some more because it can lead to some imbalanced behavior like this that's not necessarily obvious (but is working as designed).

from azure-sdk-for-go.