[jira] [Commented] (KAFKA-4713) Processors cannot call public methods on ProcessorContext from the init method

["Guozhang Wang (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-4713?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16141948#comment-16141948 ] 

Guozhang Wang commented on KAFKA-4713:
--------------------------------------

Hello [~bfelaco], thanks for reporting this issue.

The root of this is that, when `init()` is called, the processing has not been started yet, it is for initializing the processor topology graph. And at that time since no data is received, we cannot determine its record-related metadata like topic / partition / offset / timestamp. Plus a single processor can takes in data from multiple topic partitions during runtime, not only one topic-partition, so we cannot return a single topic / partition even if we know what topic partitions are assigned.

For your specific use case, could you explain a bit why your initialization logic depends on what specific topic-partition was assigned? I'd like to help finding you a workaround approach if your use case is indeed necessary.

> Processors cannot call public methods on ProcessorContext from the init method
> ------------------------------------------------------------------------------
>
>                 Key: KAFKA-4713
>                 URL: https://issues.apache.org/jira/browse/KAFKA-4713
>             Project: Kafka
>          Issue Type: Improvement
>          Components: streams
>    Affects Versions: 0.10.0.1
>            Reporter: Bruno Felaco
>
> Streams passes the ProcessorContext to the init method of a Processor on startup when a topic/partition is assigned. This is the only means that a Processor has to obtain a reference to the ProcessorContext, which is required if the implementation needs to be aware of the particular topic/partition that is being processed. However, from within the init() method, the only thing the implementation is allowed to do with the ProcessorContext is schedule periodic tasks and create/access any state stores.  It is not allowed to get the topic, partition, or current offset. This severely limits what an implementation can do in init. An implementation may need to check with or notify another component or system that it has been assigned a particular topic/partition. As far as I can see, there is no other mechanism in Streams to allow an implementation to react to the assignment of topic/partition and perform any such initialization, since Streams explicitly controls the ConsumerRebalanceListener. The implementation would have to wait for the receipt of the first record to get this information which is too late for certain use cases.
> I propose that the following methods should be usable from the init method:
> topic()
> partition()
> and also, but not as critical:
> offset()
> timestamp()
> Alternatively, streams could allow an implementation to register a separate listener that would be invoked from the global ConsumerRebalanceListener  when a topic/partition is assigned.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)