Bad behaviour when partitioning messages by key.

### Config
`application:set_env(ekaf, ekaf_partition_strategy, [{ekaf_partition_strategy, custom}])`

All messages are produced with a key.

### What happens:

1. [ekaf_server.erl:439](https://github.com/helpshift/ekaf/blob/745b2694e0c65f42239be20ade76642a388bdbaa/src/ekaf_server.erl#L439) `handle_info({worker, up, WorkerUp, WorkerUpStateName, WorkerUpState, _}, StateName, #ekaf_server { topic = Topic, messages = OfflineMessages } = State)`
2. [ekaf_server_lib.erl:118](https://github.com/helpshift/ekaf/blob/745b2694e0c65f42239be20ade76642a388bdbaa/src/ekaf_server_lib.erl#L118) `send_messages(StateName, #ekaf_server{ topic = Topic } = State, Messages)`
3. [ekaf.erl:66](https://github.com/helpshift/ekaf/blob/745b2694e0c65f42239be20ade76642a388bdbaa/src/ekaf.erl#L66) `produce_async_batched(Topic, Data)`
4. [ekaf_lib.erl:79](https://github.com/helpshift/ekaf/blob/745b2694e0c65f42239be20ade76642a388bdbaa/src/ekaf_lib.erl#L79) `common_async(Event, Topic, [{Key,Data}|Rest])`

The code waits for a reply from the `TopicWorker`, but the current process *is* the topic worker. The code *would* deadlock if we used `gen_fsm:sync_send_all_state_event` as in https://github.com/helpshift/ekaf/pull/50 . Here instead, we [send `{pick, {Key,Data}, self()}` to `self()`](https://github.com/helpshift/ekaf/blob/745b2694e0c65f42239be20ade76642a388bdbaa/src/ekaf_lib.erl#L78), immediately [`receive` it in clause `_E`](https://github.com/helpshift/ekaf/blob/745b2694e0c65f42239be20ade76642a388bdbaa/src/ekaf_lib.erl#L90) and the message is discarded.