Kafka events order and ES Process issues

[maxceem]

A big thanks to @eisbilir for catching and providing details for this issue:

Issue details

When we create a RB with dates, these ES related events will occur at taas-api :

1. Send RB creation event to bus-api
2. Send WP creation events to bus-api
   These events will always be fired with this order.
   Kafka guarantees the order of data at the partition level. The producer sends the messages in a specific order, the broker writes to the partition in that order, and the consumers read the data in that order.
   According to this logic we expect the ES processor to receive these messages in the same order.
   Even though we use 1 partition, sometimes WP creation message comes before RB's creation message.
   These messages will run sequentially because we have 1 partition. And WP creation will fail because RB hasn't been created yet.
   This is rare but happens. When I read the timestamp of the message, It's clear that the RB event was fired first. But the ES processor received them in reverse order.
   Well, I suspect 2 possibilities;
3. Even though we have 1 partition per topic, these events are related to two different topics. So, it might not be guaranteed to receive messages in the same order when they come from different topics.
4. Bus api wrapper. https://github.com/topcoder-platform/tc-bus-api-wrapper/blob/f8cbd335a0e0b4d6edd7cae859473593271fd97f/src/common/helper.js#L26 Is it possible that when waiting for m2m token, the order is broken due to network issues?
   I am not sure, more debugging and log reading is needed.
   I just wanted to share this experience with you, because i think it would be good to have this info for future bugs, because this scenario happens with a probability of 1/20~30 for me.

Issue reason

This happens because the order of messages from different topics is not guaranteed by Kafka, so even if we send them in the correct order there is no guarantee they would come in that order in the consumer.

[maxceem]

I guess the only solution which is to merge all the create topics into one (and same for update and delete) the same way as it's done in submission-es-processor and project-processor-es.

This way our topic naming would also follow Topcoder Standard for Kafka Topics naming.

The general idea is the next one, instead of having a separate topic for each kind of resources like job, resource booking, job candiatate, interview and so on. We would have only general topic for each operation create, update, deleted and bulkUpdate. And the type of the resource should be passed inside the message in the fields resource.

So all the current topics should be converted:

Current Topic	New Topic	Resource
Job
taas.job.create	taas.action.create	job
taas.job.update	taas.action.update	job
taas.job.delete	taas.action.delete	job
Job Candidate
taas.jobcandidate.create	taas.action.create	jobcandidate
taas.jobcandidate.update	taas.action.update	jobcandidate
taas.jobcandidate.delete	taas.action.delete	jobcandidate
Resource Booking
taas.resourcebooking.create	taas.action.create	resourcebooking
taas.resourcebooking.update	taas.action.update	resourcebooking
taas.resourcebooking.delete	taas.action.delete	resourcebooking
Work Period
taas.workperiod.create	taas.action.create	workperiod
taas.workperiod.update	taas.action.update	workperiod
taas.workperiod.delete	taas.action.delete	workperiod
Work Period Payment
taas.workperiodpayment.create	taas.action.create	workperiodpayment
taas.workperiodpayment.update	taas.action.update	workperiodpayment
Interview
taas.interview.requested	taas.action.create	interview
taas.interview.update	taas.action.update	interview
taas.interview.bulkUpdate	taas.action.bulkUpdate	interview
Interview
taas.role.requested	taas.action.create	role
taas.role.update	taas.action.update	role
taas.role.delete	taas.action.delete	role

[maxceem]

How this issue happens:

• Inside ES we have index with stores Resource Bookings and inside each of the Resource Bookings we store Work Periods as nested objects, because we want to filter by the complex queries:

  {
    "id": "resource booking id",
    // work periods as nested objects
    "workPeriods": [
        { "id": "work period id 1" },
        { "id": "work period id 2" },
        // ...
        { "id": "work period id N" }
     ]
  }

• We send request to TaaS API to create a Resource Bookings:
  • Resource Bookings is created and event taas.resourcebooking.create is sent to Kafka
  • automatically for Resource Bookings we create Work Periods inside TaaS API, and send events taas.workperiod.create for each created Work Period
• These events are in different topics, thus Kafka doesn't guarantee that these events would be consumed in the same order as they fired.
  • So this might happen that taas.workperiod.create event is processed first and it causes error, as Resource Bookings is not there yet, and we cannot create Work Period inside Resource Booking as Resource Booking is not yet created.

[maxceem]

The approach with changing event naming structure would affect a lot of feature and very dangerous to implement as there is a lot of work begin done in this repo in parallel. So we can try to find some local solution for our particular issue.

For example we can try like this:

• everytime when we get create WP event we check if RB exists, if doesn't exists, we try a few more times after 500ms, 1 second, and 3 seconds, if RB is still not there, only then we throw error

@eisbilir what do you think? would this work?

[eisbilir]

@maxceem I agree with your 2nd solution.
Events other than those related to RB only lead to partially updates. Co-creation of parent and child only happens for RB.
So I think it's better to focus on some local solutions. Retrying with delay on failures would work.

[maxceem]

We've also discussed it internally and come up with the next approach.
When WP is failed to be created because RB doesn't exist yet it in ES we have to repost event to Kafka using a special topic taas.action.retry with the next payload:

payload: {
   originalTopic: 'taas.workperiod.create',
   originalPayload: { ... }, // original message payload, for WP it would be the Work Period object
   retry: 1, // the number of retry   
}

We should send this message to the Kafka using Bus API, similar to how we do from TaaS API (create reusable method postEvent).

Before sending we have to make a delay using "exponential backoff", i. e. delayBeforeRetry = 2^retry * BASE_RETRY_DELAY. Where BASE_RETRY_DELAY should be configurable using env variable. 250 ms for now.

And we should make not more than MAX_RETRY which also should be configurable and should be equal to 3 for now.

If the processing retry topic fails (because of RB doesn't exists), we have to send retry message again with increased retry count.

It's very important that we send retry topic only if RB doesn't exists, and don't send it if another error happens. I think it could be detected by error.statusCode === 404

For now let's only apply this logic to taas.workperiod.create topic, but it would be great if potentially we could easily apply the same logic to other topics.

Add this topic to local Kafka config in ES Processor and in TaaS API repositories.

This task should be mainly implemented in the ES Processor repo, branch dev.

[maxceem]

@eisbilir would you like to take care of this issue as per description in the comment above? #296 (comment)

Payment: $150. (payment increased because you also localized and provided details for this issue)

If you have any suggestions on how to improve the suggested way to implement, please, let me know.

NOTE: if it's hard to implement several retries, or it requires many changes, we can simplify the task to only one retry after a fixed delay. So we would like to support several retires, if we can keep the code clean and avoid complexity. It would be really fine to simplify it to one retry, just let me know how you think is better to keep logic clean.

[eisbilir]

Ok, I'll take care of this, and the solution #296 (comment) looks perfect.

I don't think it would make it complex to implement several retries and It'd be easily applied any other topic when needed.

[maxceem]

One more thing. When we send Kafka message we set originator which indicates service which sent this message, see https://github.com/topcoder-platform/taas-apis/blob/dev/src/common/helper.js#L944.

We have to send retry messages from ES Processor by setting originator = taas-es-processor. And inside the processor we should only process retry event if originator is taas-es-processor. This is needed if we have another processor which also implements retry logic, so they should only process retry messages which they sent.

[eisbilir]

PR #306
topcoder-platform/taas-es-processor#69