SegmentAllocateAction (fixes #1515)

[gianm]

This is a feature meant to allow realtime tasks to work without being told upfront
what shardSpec they should use (so we can potentially publish a variable number
of segments per interval).

The idea is that there is a "pendingSegments" table in the metadata store that
tracks allocated segments. Each one has a segment id (the same segment id we know
and love) and is also part of a sequence.

The sequences are an idea from @cheddar that offers a way of doing replication.
If there are N tasks reading exactly the same data with exactly the same logic
(think Kafka tasks reading a fixed range of offsets) then you can place them
in the same sequence, and they will generate the same sequence of segments.

[himanshug]

s/pendingSegmentTable/pendingSegmentsTable just for consistency

[gianm]

sure, this looks like a typo

[himanshug]

say a task T allocates a new segment and hands it off, then it receives another event in the same interval. This time, task is supposed to use the "previousId" so that a new segment gets allocated. For this, is task supposed to retain a list of all the segments it has handed off in its lifetime? What happens if/when task is restored (do we keep that list maintained on disk as well) ?

[gianm]

@himanshug it needs to remember the last segment that has been allocated in its sequence (last overall, not last per interval, so just one segment).

[gianm]

bad merge from master, will fix shortly.

[himanshug]

@gianm I'm not sure how following case works.
say T1 and T2 are replicated tasks (with same group id)
T1 receives event e, allocates a segment-id for same and hands it off at some point
T2 is slow and now receives event e, will it get a new segment-id and hand it off successfully resulting in duplication of data?

[himanshug]

@gianm trying to ans my own question. I guess T2 will receive same segment-id that T1 "created" and will eventually try to hand-off the same segment but will fail because commit metadata wouldn't match.

[himanshug]

note: this assumes, tasks in different replication set "share" the lock so that they both will be able to get lock on same interval if they received events with same or close enough timestamp.

[gianm]

@himanshug yes, that's what would happen, T1/T2 get the same segment id because they see the same events in the same order, and they're allocating ids from the same sequence.

IMO in that case the slow task shouldn't fail- it should realize that the faster task handed off that segment first (based on inspection of the commit metadata when its attempt to commit fails) and either keep continuing on or skip forwards.

[gianm]

Tests run: 7, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 600.39 sec <<< FAILURE! - in io.druid.server.namespace.cache.NamespaceExtractionCacheManagerExecutorsTest
testConcurrentAddDelete(io.druid.server.namespace.cache.NamespaceExtractionCacheManagerExecutorsTest)  Time elapsed: 600.007 sec  <<< ERROR!
java.lang.Exception: test timed out after 600000 milliseconds
    at sun.misc.Unsafe.park(Native Method)
    at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
    at java.util.concurrent.FutureTask.awaitDone(FutureTask.java:429)
    at java.util.concurrent.FutureTask.get(FutureTask.java:191)
    at io.druid.server.namespace.cache.NamespaceExtractionCacheManagerExecutorsTest.testConcurrentAddDelete(NamespaceExtractionCacheManagerExecutorsTest.java:312)

[himanshug]

👍

[fjy]

👍

[drcrallen]

@gianm this causes lots of retries for duplicate primary key entry errors that will never be transient unless someone manually cleans the DB

[gianm]

#2573 should fix this