migrate event handlers to one generic handler

[jayshrivastava]

This change migrates this pattern:

SessionStateBuilder::new()
    .with_distributed_event_handler(desired_task_count)
    .with_distributed_event_handler(scale_up_leaf_node)
    .with_distributed_event_handler(route_tasks);

to this:

SessionStateBuilder::new()
    .with_distributed_event_handler(desired_task_count)
    .with_distributed_event_handler(scale_up_leaf_node)
    .with_distributed_event_handler(route_tasks);

The type of handler can be inferred by the argument. Most of the
relevant change are in src/events.

[jayshrivastava]

Any usize automatically routes to EventHandler<DesiredTaskCountHandler> which is kind of funny.

One day someone could implement impl EventHandler<MyNewKindOfHandler> for usize which means rust cannot resolve with_distributed_event_handler(2). You would have to do something like this to get it to work with_distributed_event_handler::<DesiredTaskCountHandler>(desired_task_count)

[gabotechs]

🤔 This... does not look very good. with_distributed_event_handler(2) is too ambiguous. This implementation is only used for tests, if you ask me, I'd just remove it then.

[gabotechs]

🤔 This... does not look very good. with_distributed_event_handler(2) is too ambiguous. This implementation is only used for tests, if you ask me, I'd just remove it then.

[gabotechs]

There's a challenge with this shape. Some other event handlers, do not need to return an Option<>, for example: https://github.com/datafusion-contrib/datafusion-distributed/pull/568/changes#diff-7bd36235787107a21e71a52b4e5511fb7c3fb66c4c6180820a6712c78022c3fe.

One thing that sounds cornering is the fact that all EventHandlers need to adhere to the same shape.

Another example is making the DesiredTaskCountHandler async:

• Make desired task-count handlers async #569

We'd like that event hander to be async, but we want to keep the other handlers synchronous, however, I imagine here either all of them are async or none are.

[jayshrivastava]

Initially, I had

fn handle(&self, ev: E::Data<'_>) -> E::Response;

but then that means each Handler needs to implement its own find_map behavior. It is nice having it here in one spot.

impl<E: Event> EventHandlerChain<E> {
    pub(crate) fn handle(&self, ev: E::Data<'_>) -> Option<E::Response> {
        // Give priority to custom handlers registered by users.
        if let Some(res) = self
            .custom
            .iter()
            .find_map(|handler| handler.handle(ev.clone()))
        {
            return Some(res);
        }
        // If no user handler handled the event, use the built ins.
        self.builtin
            .iter()
            .find_map(|handler| handler.handle(ev.clone()))
    }
}

The async thing is a bigger problem. Let me think about this

[jayshrivastava]

Well we could have two traits.
EventHandler and AsyncEventHandler