Enable partition stats on streaming task completion report

[ac9817]

Description

This adds visibility into number of partitions processed by each streaming task and the count of records processed from each of those partitions.

This PR updates IngestionStatsAndErrorsTaskReportData to take new map which represent count by partition and SeekableStreamIndexTaskRunner populates that map while processing each record. We are trying to analyze the work done by each task so as to efficiently assign appropriate resources in future to the worker nodes.

---

Release note

Streaming Task completion report now has an extra field recordsProcessed which lists all the partitions processed by that task and count of records for each partition. One can look at this field to see the actual throughput of tasks and make decision as to vertically/horizontally scale there workers.

---

Key changed/added classes in this PR

• SeekableStreamIndexTaskRunner.java
• IngestionStatsAndErrorsTaskReportData.java

---

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[kfaraz]

Thanks for the PR, @adithyachakilam!

Overall, the change makes sense to me. Could you add some info of why you needed this stat and if it helped you debug any failure scenario?

[suneet-s]

Thanks for your first contribution @adithyachakilam ! I have one suggestion on naming the object in the task report. Kashif also has some good suggestions to improve code readability.

LGTM after the rename for the object in the task report and associated docs.

In the PR description, can you update the Release note section to answer why someone may want to read the recordsProcessed field from the task report.

[ac9817]

Thanks for the comments @kfaraz / @suneet-s. Tried to address them, let me know if you need anything more.

[kfaraz]

LGTM, +1 after CI passes.

[ac9817]

@kfaraz seems like an unrelated flaky test that has failed.

[kfaraz]

Thanks for your first contribution, @adithyachakilam !
Merging the PR as all comments have been addressed and the CI failure is due to unrelated failing integration test.

[kfaraz]

I have already merged the PR but taking another look at the final payload, I feel recordsProcessedPerPartition or partitionRecordsProcessed might be a more appropriate name. Right now the payload would look something like:

"recordsProcessed": {
   "1": 1234,
   "2": 2345
}

Looking at this payload, it is difficult to tell exactly what each entry in the map represents.

@suneet-s , @adithyachakilam , what do you think?

[ac9817]

@kfaraz makes sense, made the change here: #15958