[Question] KafkaIO Sink/Source

[kevinsookocheff-wf]

Hi, I've been working on a streaming KafkaIO.Read and KafkaIO.Write. From this conversation it seems like Google is already working on this and I don't want to duplicate effort. Can anyone comment on the status of this? I'd be happy to help get it out the door.

http://stackoverflow.com/questions/29893342/is-it-possible-to-use-kafka-with-google-cloud-dataflow

[dhalperi]

@kevinsookocheff-wf: We are not working actively on Kafka connectors at this time. I would be happy to help review your code or answer questions as your implementation moves forward.

[kevinsookocheff-wf]

I started working on this using the new consumer API available in Kafka 0.9. One question I had was which pattern to use to implement this. I could inherit from the abstract Source classes as described here. I also see a lot of the standard IO features following a different pattern using the Bound class. The contrib modules currently in pull request mix both styles. Ultimately, to me it doesn't make much difference, I would just like to know what the recommended style is.

I am also interested in having both unbounded source and sinks and the abstract sink classes don't exist yet for the unbounded case.

[dhalperi]

Indeed, our IO connectors do not have a consistent style: the guidelines for Custom IO you linked came later and are more refined. We do recommend that you follow the Custom IO guidelines for your sources and sinks.

We would like to provide a good abstraction for custom unbounded sinks, but you're right that this is not available right now.

[kevinsookocheff-wf]

@dhalperi I've made a lot of progress on this but there are some parts of the custom source API that I'm not sure I understand with the current documentation. Particularly around failure scenarios.

What I've done so far is read messages from Kafka into a buffer to be processed with the Dataflow SDK. These messages have been acknowledged to Kafka as being read but I need a way to handle the case where I have acknowledged the messages but the Dataflow process fails and restarts before I'm able to process the messages. Is there a point in the UnboundedSource API that I can serialize unprocessed data and restore it when the service comes back up.

[dhalperi]

Hi @kevinsookocheff-wf -- great to hear that you're making progress!

We expect that you will use checkpointing (see getCheckpointMark and other related functions) to include state you don't want to lose. The runner itself will call getCheckpointMark to ensure that data is persisted.

Note that this case is explicitly called out here:

getCheckpointMark: The Dataflow service uses this method to create a checkpoint in your data stream. The checkpoint represents the UnboundedReader's progress, which can be used for failure recovery. Different data streams may use different checkpointing methods; some sources might require received records to be acknowledged, while others might use positional checkpointing. You'll need to tailor this method to the most appropriate checkpointing scheme. For example, you might have this method return the most recently acked record(s).

[dhalperi]

Also note that you may want to hold the acks back until the finalizeCheckpoint method is called: https://github.com/GoogleCloudPlatform/DataflowJavaSDK/blob/master/sdk/src/main/java/com/google/cloud/dataflow/sdk/io/UnboundedSource.java#L104

[dhalperi]

Hi Kevin,

Sounds like this particular issue has been resolved. Please reopen/open a new issue for the next one :).

Thanks,
Dan

[nicolasembleton]

Was the connector ever finished? @kevinsookocheff-wf have you published it? Was trying to look for it but can't seem to see it.

[fsalem]

Hi,

I'm trying to implement UnboundedSource reading from Kafka. When I run it, this exception is thrown "no evaluator registered for UnboundedSource [Read(UnboundedSource)]"

After debugging in the source code, I discovered that there is a difference between "Bound" class and "Unbound" class inside "Read" class. The "Bound" class registers a new transform evaluator while "Unbounded" class doesn't.

Could you please advise ? Am I right? or I miss something ?

Best regards,
Farouk Salem

[kevinsookocheff-wf]

I have something that would be considered a prototype. We've moved in a different direction (off Kafka) so it was never fully tested. I will see if it's possible to release what I have that could be used as a basis for future work.

[dhalperi]

Hi Farouk,

Please open a new issue for your problem. Also, please include a brief code outline or a gist for how you defined your unbounded source.

I would not expect to see that error. You may wish to read the Dataflow web docs on Custom Sources and Sinks.

Thanks!
Dan

[big-andy-coates]

@kevinsookocheff-wf that would be much appreciated :D