Search code examples
javahazelcasthazelcast-jet

Creating a new Jet custom Partitioner

My use case requires to read messages from a Kafka topics and process the messages in the natural order as they were published into the Kafka.

The Kafka producer is responsible to publish each group of messages sorted in a single kafka topic-partition, and I need to process each group of message in the same Vertex-Processor in the same order.

enter image description here

The image above represents the basic idea. There a few KafkaSource-Processors reading from Kafka.

And one edge connected to a vertex to decode the kafka message and so on.

I could use the kafka message key as the partitioning key, but I think that I will end up with unbalanced decode processor.

Given that:

  • How can I create a new Partitioner ? I couldn't find any example to inspire me.
  • On the new Partitioner, how can I identify KS processor that emitted the message ? I would like to have a 1-to-1 relationship between previous vertex process and the next vertex processor, for instance, KS#0 always send the messages to the Decode#0, KS#1 to Decode#1 and so on.
  • Do I need a new partitioner for that or is there some out-of-the-box functionality to achieve that ?
Solution

  • You don't need to use partitioner for this. Edge.isolated() together with equal local parallelism is designed for this:

    dag.edge(between(kafkaSource, decode).isolated());

    In this case, one instance of source processor is bound with exactly one instance of target processor and ordering of items will be preserved. Keep in mind that single Kafka source processor can take items from more than one Kafka partition, so you have to track the Kafka partition id. Even if you make the total number of Jet processors and Kafka partitions equal, you can't rely on it, because if one of the members fails and the job is restarted, the total number Jet processors will decrease but the number of Kafka partitions won't.

    Also note that default local parallelism is not equal for sources: For Kafka source it defaults to 2, for others it typically is equal to CPU count. You need to manually specify equal value.

    Another limitation is if you use Processors.mapP for your decode vertex, the mapping function must be stateless. Because you need the items to be ordered I assume that you have some state to keep. For it to work correctly, you have to use custom processor:

    Vertex decode = dag.newVertex("decode", MyDecodeP::new);

    Processor implementation:

    private static class MyDecodeP extends AbstractProcessor {
    private Object myStateObject;

    @Override
    protected boolean tryProcess(int ordinal, @Nonnull Object item) {
        Object mappedItem = ...;
        return tryEmit(mappedItem);
    }
}

    The answer was written for Jet 0.5.1.