Comments (6)
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:
- Are you seeing duplicate messages?
- Are you seeing any failures from partitionClient.UpdateCheckpoint? If that function fails restarts will end up consuming the same events again.
- Are you running multiple Processor instances? And if so, are all of those instances using the same consumer group?
- 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.
- No
- No
- 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).
- 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.
Related Issues (20)
- Support update container registry cache rule source repository and target repository HOT 2
- [Question] How does token Caching work HOT 5
- [azeventhubs] NextPartitionClient does not always respect context cancellation HOT 3
- [azcosmos] 1.0.3 - The provided cross partition query can not be directly served by the gateway HOT 4
- Question: How can I count the number of blobs in a storage container? HOT 3
- DownloadFile without Range.Count incurs 2 calls to GetProperties HOT 2
- Update the doc of InteractiveBrowserCred and DeviceCodeCred
- Pagination of Availability Statuses does not work HOT 5
- Go SDK for commercial-marketplace-client HOT 1
- Clean up track1 tool in eng/tools
- service Package HOT 1
- Download a specific version from Azure blob HOT 3
- Allow expiring authorization token for 403 response HOT 7
- Allow passing any policy on individual request done with a client HOT 1
- Cloud Shell ManagedIdentitySource should throw if a user-assigned managed identity clientID or resourceID is supplied
- Port AuthenticationRecord related caching doc HOT 1
- [Identity] AzurePipelinesCredential parameterless constructor and add to DAC HOT 1
- [Identity] Create a new credential to support FIC in Github Actions HOT 1
- Add tracing to the Cosmos Client HOT 7
- ARM Resource Health NewEmergingIssuesClient().NewListPager NextPage is failing to unmarshall armresourcehealth.EmergingIssue.RefreshTimestamp HOT 5
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from azure-sdk-for-go.