Read kafka offsets from checkpoint - state processor

[bat man]

Hi Users,

Is there a way that Flink 1.9 the checkpointed data can be read using the state processor api. 

Docs [1] says - When reading operator state, users specify the operator uid, the state name, and the type information.

What is the type for the kafka operator, which needs to be specified while reading the state.

[1] - https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/libs/state_processor_api/

Thanks,

Hemant

[bat man]

Anyone who has tried this or can help on this.

Thanks.

On Thu, May 6, 2021 at 10:34 AM bat man <[hidden email]> wrote:

Hi Users,

Is there a way that Flink 1.9 the checkpointed data can be read using the state processor api. 

Docs [1] says - When reading operator state, users specify the operator uid, the state name, and the type information.

What is the type for the kafka operator, which needs to be specified while reading the state.

[1] - https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/libs/state_processor_api/

Thanks,

Hemant

[Dawid Wysakowicz-2]

Hi,

You would need to look into the internals of FlinkKafkaConsumerBase. In the current master the state for offsets is initialized in here: https://github.com/apache/flink/blob/fbf84acf63102db455c89cb8e497cda423a1c4d5/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java#L984

Hope it helps.

Best,

Dawid

On 07/05/2021 14:43, bat man wrote:

Anyone who has tried this or can help on this.

Thanks.

On Thu, May 6, 2021 at 10:34 AM bat man <[hidden email]> wrote:

Hi Users,

Is there a way that Flink 1.9 the checkpointed data can be read using the state processor api. 

Docs [1] says - When reading operator state, users specify the operator uid, the state name, and the type information.

What is the type for the kafka operator, which needs to be specified while reading the state.

[1] - https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/libs/state_processor_api/

Thanks,

Hemant