SAMZA-2444: JobModel save in CoordinatorStreamStore resulting flush for each message

[alnzng]

Symptom

When Samza's job creates lots of tasks/partitions, it can take over a long time for AM to save the metadata in a run which may cause timeout exception.
We observed if the Samza's job has over 37k tasks/partitions, it takes over 10 min for AM to save it in a run.

Cause

JobModelManager uses CoordinatorStreamStore.put() to save job metadata information which does flush for each message, and the flush operation is heavy especially when the remote server suffering the performance issues.

Changes

• Separate flush from put/putAll/delete functions in CoordinatorStreamStore.
• Explicity call flush after call put/putAll/delete in related classes
• Batch write task partition assignments information to metadata store.
• Batch write task container information to metadata store.

Tests

• All unit tests and integration tests are passed

API Changes

• Replace writeTaskPartitionAssignment with new batch write method writeTaskPartitionAssignments in TaskPartitionAssignmentManager
• Replace writeTaskContainerMapping with new batch write method writeTaskContainerMappings in TaskAssignmentManager.

Upgrade Instructions

None

Usage Instructions

None

[prateekm]

@bharathkk and @lakshmi-manasa-g can you take a look?

[xinyuiscool]

Now I looked at the coodinator stream impl for metadatstore, it seems we shouldn't couple flush with put/putall/delete/deleteall. It should be done as a separate call at the end of updates. The current impl is hideous and can cause further perf problems down the road.

[alnzng]

Now I looked at the coodinator stream impl for metadatstore, it seems we shouldn't couple flush with put/putall/delete/deleteall. It should be done as a separate call at the end of updates. The current impl is hideous and can cause further perf problems down the road.

@xinyuiscool Do you mean we should remove flush calls from put/putall/delete/deleteall? And let the callers of those functions decide when to do flush operation?

[xinyuiscool]

Yes, I think that's a typical store api. Not sure why it was doing all the flushes before. @bharathkk might have some context about it, but he is on vacation.

[alnzng]

@lakshmi-manasa-g Thanks for the review. Had one discussion with @xinyuiscool and @prateekm , we decided to change with another solution which will move the flush out of all update methods in CoordinatorStreamStore. So above codes with your comments will be removed. I will let you know when the new changes are done.

[prateekm]

Alan, this PR seems to be solving a similar problem. #1125

Can you check how the problem / solution in that PR relates to this one?

[alnzng]

Alan, this PR seems to be solving a similar problem. #1125

Can you check how the problem / solution in that PR relates to this one?

@prateekm
Just checked the PR: #1125. I believe it is solving exactly the same performance issue with this one.
But it is doing with batch updates way, and current PR is doing the way to decouple flush and update methods.

If we solve the issue with this PR, later we can close it.

[alnzng]

The changes are done and updated the details in the PR's description, please help review, thanks!
@xinyuiscool @prateekm @lakshmi-manasa-g

[mynameborat]

Yes, I think that's a typical store api. Not sure why it was doing all the flushes before. @bharathkk might have some context about it, but he is on vacation.

@alnzng here is the PR for context - #1112
+1 to splitting flush and the API. Refer to the comments in the above PR regarding separating out flush and put APIs

[lakshmi-manasa-g]

thank you for addressing the comments.

[alnzng]

@xinyuiscool @prateekm
Can any of you help check and merge the PR if no new comments? Thanks.

[xinyuiscool]

LGTM. Thanks for the quick fix!