add control and status endpoints to KafkaIndexTask

[dclim]

Added the following HTTP endpoints to KafkaIndexTask:

1. [GET /offsets/current] - Return which offsets we are currently processing
2. [GET /offsets/end] - Return the offsets that the task will read until
3. [POST /offsets/end] - Modify the ending offsets to allow the task to continue reading or stop early
4. [POST /stop?publish=true/false] - Signal the task to stop early, either abruptly or gracefully (stop reading and publish segment with what we currently have). In both cases it returns a SUCCESS status
5. [POST /pause?timeout=x] - Signal the task to pause reading, either indefinitely or with a provided timeout to resume
6. [GET /status] - Return the task's state
7. [POST /resume] - Signal the task to resume reading
8. [GET /time/start] - Returns the timestamp when the task began running

These endpoints are required for implementing functionality for #2656, specifically:
(1) exposes the offset currently being read and allows us to start the next tasks early
(4) allows us to stop redundant replicas when another has completed without generating a failed status
(all) allows us to stop tasks gracefully in a coordinated way across replicas (by pausing and then setting the end offsets so that they all finish at the same place) - this enables graceful immediate schema rollover as well as time-based task lifetimes instead of number of event based
(all) in conjunction with a pauseAfterRead flag in KafkaIOConfig, provides a framework to enable leader/follower capabilities, where followers will pause when they are done reading their offsets until the leader assigns a new set of ending offsets

[himanshug]

nit: following is more desired in java land if you wanted to iterate through both keys and values in a map

 for (Map.Entry<Integer, Long> entry : offsets.entrySet())

[dclim]

sounds good, thanks!

[himanshug]

👍 after #2730 (comment)

[gianm]

This doesn't look like it does what a stopGracefully should do. Maybe we should rename that method to something like suspend (not in this PR though). But, what it's supposed to do is suspend a task quickly so the jvm can be shut down and then restarted. Specifically it shouldn't be trying to publish or anything like that – just persist to disk and then stop.

[dclim]

I can swap the implementations of stopGracefully and stopAbruptly since the abrupt one does what you described.

[dclim]

@gianm thanks for the comments, I added some follow-up questions/thoughts

[dclim]

@gianm addressed your comments

[gianm]

status should start at something – perhaps NOT_STARTED or some such

[dclim]

sounds good

[gianm]

How is status kept consistent? Are all writes going through a single thread, or are they protected by a lock? Would be good to have that in a comment.

[dclim]

Yes, status is only modified by the task runner thread which calls run(). I added a comment.

[gianm]

👍 after #2730 (comment)

[himanshug]

why we stop reading altogether after one partition was done and pauseAfterRead was set? wouldn't this terminate the task earlier than we wanted.

[dclim]

If pauseAfterRead is true, stillReading will remain true regardless of the state of assignment - this will ensure we continue to loop and will pause in possiblyPause() if all partitions have been read. If pauseAfterRead is false, the loop will end as soon as all the partitions have been read.

[himanshug]

ok, so pauseAfterRead ensures that task does not stop even after completion of consuming all partitions upto given offsets... and possibly supervisor would make requests to stop and publish later.

[himanshug]

👍 besides #2730 (comment)

[dclim]

Thanks!