Adding key object when creating a SourceRecord in Kafka Source Connector

[Yeshwanth Kumar]

Hello all, 

I am currently playing around a bit with the stream-reactor cassandra source connector. For a given set of events in cassandra, I want to make one of the columns as the partition key when inserting the record into kafka. 
Kafka's default partitioner will do the work of choosing the right partition provided I give the right key.

Currently, it doesn't take partition no. or key when creating a SourceRecord, and hence I believe the partitions will be chosen in a round robin fashion. 

https://github.com/datamountaineer/stream-reactor/blob/master/kafka-connect-cassandra/src/main/scala/com/datamountaineer/streamreactor/connect/cassandra/source/CassandraTableReader.scala#L270

I added a keyObject and keySchema,

`new SourceRecord(sourcePartition, Map(timestampCol -> offset), topic, keystruct.schema(), keyStruct, struct.schema(), struct)`

I do create a key object by ignoring all column except for the key's column in cassandra. It still does not push it to multiple partitions inspite of having more than 1 partition. 
Am I missing something?

Thanks,
Yesh

[Ewen Cheslack-Postava]

You're not missing anything, that is expected to work -- the key gets serialized and used as the Kafka key, so it should work just like a normal producer. Have you tried dumping the data with console consumer and verifying that it has keys and that they are not all identical?

-Ewen

--
You received this message because you are subscribed to the Google Groups "Confluent Platform" group.
To unsubscribe from this group and stop receiving emails from it, send an email to confluent-platform+unsub...@googlegroups.com.
To post to this group, send email to confluent-platform@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/confluent-platform/534aaac4-9357-4d2f-a85c-69ceb0c4bf88%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Yeshwanth Kumar]

yes, it worked. But I need to have a minimum of 5 partitions for two possible keys. 

On Wednesday, May 3, 2017 at 5:16:34 PM UTC+1, Ewen Cheslack-Postava wrote:

You're not missing anything, that is expected to work -- the key gets serialized and used as the Kafka key, so it should work just like a normal producer. Have you tried dumping the data with console consumer and verifying that it has keys and that they are not all identical?

-Ewen

On Wed, May 3, 2017 at 7:58 AM, Yeshwanth Kumar <morph...@gmail.com> wrote:

Hello all, 

I am currently playing around a bit with the stream-reactor cassandra source connector. For a given set of events in cassandra, I want to make one of the columns as the partition key when inserting the record into kafka. 
Kafka's default partitioner will do the work of choosing the right partition provided I give the right key.

Currently, it doesn't take partition no. or key when creating a SourceRecord, and hence I believe the partitions will be chosen in a round robin fashion. 

https://github.com/datamountaineer/stream-reactor/blob/master/kafka-connect-cassandra/src/main/scala/com/datamountaineer/streamreactor/connect/cassandra/source/CassandraTableReader.scala#L270

I added a keyObject and keySchema,

`new SourceRecord(sourcePartition, Map(timestampCol -> offset), topic, keystruct.schema(), keyStruct, struct.schema(), struct)`

I do create a key object by ignoring all column except for the key's column in cassandra. It still does not push it to multiple partitions inspite of having more than 1 partition. 
Am I missing something?

Thanks,
Yesh

--
You received this message because you are subscribed to the Google Groups "Confluent Platform" group.
To unsubscribe from this group and stop receiving emails from it, send an email to confluent-platform+unsub...@googlegroups.com.

To post to this group, send email to confluent...@googlegroups.com.