PDP 51 (Segment Container Event Processor) - derekm/pravega GitHub Wiki

Table of Contents:

Motivation

We need to implement a durable queue inside each Segment Container. This should allow internal components (such as SLTS) to queue up delayed/lazy tasks that need not be executed right away but should be executed at one point in the future (ex: chunk cleanup tasks).

Requirements

These are the initial requirements of the Segment Container Event Processor, as described in Issue 5909:

We should add a new sub-service to each Segment Container. Call this an Event Processor. Each Event Processor:

  • Has exactly one consumer. When being created, a consumer name needs to be provided.
  • Is backed by a single, pinned, system-critical segment whose name is a function of the container ID and the consumer name.
  • Acts as a FIFO queue. Every item added is appended to the end of the segment. The Segment is also tailed, and when events are picked up, a consumer lambda should be invoked to handle them, after which the processed events should be truncated out of the segment.
  • This segment should have a rather small rollover size (like 4MB). We need to add proper metrics to this processor:
  • Outstanding byte size (not count)
  • Average processing latency

Design

  • We need a new interface ContainerEventProcessor that defines the methods and classes to be used in the Segment Container Event Processor service.
    • This includes the methods for ContainerEventProcessor, the EventProcessor interface and the EventProcessorConfig class.
  • SegmentContainer interface now also implements ContainerEventProcessor and StreamSegmentContainer provides the implementation for it.
  • ContainerEventProcessorImpl encapsulates the implementation of ContainerEventProcessor:
    • It uses an internal, system-critical Segment for each EventProcessor registered.
    • Each event added via EventProcessor implementation contains the length of the event size (can be variable in size) followed by the event itself. This is necessary to provide a list of individual events to the handler function in EventProcessor.
  • There is a special attribute on each internal EventProcessor Segment that durably stores the last successfully processed event (i.e., truncation point). This allows the ContainerEventProcessor service to resume processing from the right offset even after restarts.
  • ContainerEventProcessor should be also in charge of reporting metrics and ensuring that the outstanding bytes do not grow without bounds.

Assumptions

  • The internal Segment for an EventProcessor durably stores the data in the order in which it is appended.
  • If there is a critical problem related to an EventProcessor's Segment (e.g., DataCorruptionException), the Segment Container will shut down.

Limitations

  • While exceptions in the execution of an EventProcessor handler are expected, if we realize of a bug in a handler, there is no other option that fixing the bug via code change (i.e., requires re-deployment).
  • The framework needs to encode the length of individual events as a header for each add() operation, as otherwise we just read a continuous range of bytes from a Segment (i.e., there is no notion of "event"). In the same way, reads in ContainerEventProcessor need to parse the length of an event prior to actually reading the event itself.

Discarded Approaches

  • Initially, we discussed the option of re-using the same the "Controller Event Processor" framework to also cope with the requirements described here. However, while the overall goal in both cases is similar, the Controller Event Processor framework design/implementation is too coupled to the use of the Pravega Client, apart from contain many features that are not needed in the Segment Store. For this reason, we discarded refactoring this framework in a way that could serve for both Controller and Segment Store.

References