KAFKA-12754: Improve endOffsets for TaskMetadata

[wcarlson5]

Improve endOffsets for TaskMetadata also add an int test for TaskMetadata offset collections

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[wcarlson5]

@rodesai @abbccdda @ableegoldman I had to make a couple changes to the task metadata to improve when the end offset was updated. Now we get it at poll phase which should give us the highest offset streams has seen at any point

[ableegoldman]

@wcarlson5 there's a failure that I'm guessing is related: streams.integration.TaskMetadataIntegrationTest.shouldReportCorrectCommittedOffsetInformation

[wcarlson5]

@ableegoldman yep it looks like I am going to have to go back to the drawing board for either the test or the impl. I will look at it tomorrow

[ableegoldman]

I'm a little out of the loop with the Task Metadata stuff, but if my understanding of things is correct then the issue with updating the end offsets is slightly different than what's addressed here. Basically we currently piggy-back on the StreamThread's #updateThreadMetadata, but that's only invoked once when the thread goes from restoring to running. It seems like we should update the TaskMetadata separately, and more frequently, right? For example at each poll invocation. Or the TaskMetadata#endOffsets could even just call through to each StreamTask#highWatermark for a fully up-to-date view of the end offsets, though the tradeoff is of course we make that call a bit more expensive. Thoughts?

Also, unrelated to your PR or this feature, it seems like the ThreadMetadata should have been updated at the very least any time the thread state changes, since the thread state is one of the essential pieces of the thread metadata...not sure why we would only update it when it goes to RUNNING, that seems to make the state field of the metadata pretty much useless. Maybe you can address that in your PR on the side (should be a one-line change in StreamThread#setState)

[wcarlson5]

@ableegoldman Thanks for the review! I think I got to your comments. There was one miss understanding about how the TaskMetadata is updated but let me know if you still have questions about that.

[ableegoldman]

Just curious, is there a reason for producing the messages one at a time, rather than all together at the beginning and then relying on process to gate when they can be processed?

[wcarlson5]

not really, I wrote the process gate first before I wanted to have more than one at a time. I think it makes the consumer have to poll multiple times as it alternates between the pollPhase and processing but I am not sure if that actually improves the test.

[ableegoldman]

LGTM, can you just fix the error messages in the test? After that I'll go ahead and merge

[wcarlson5]

@ableegoldman I think it ready :)

[ableegoldman]

Looks like just the usual flaky test failures in the build: RaftClusterTest.testCreateClusterAndCreateAndManyTopicsWithManyPartitions() and KTableKTableForeignKeyInnerJoinMultiIntegrationTest.shouldInnerJoinMultiPartitionQueryable()