[FLINK-3045] Properly expose the key of a Kafka message

[rmetzger]

The current Kafka connector does not allow users to access the message key.

With this change, I've added a new pair of serialization schemas (KeyedDeserializationSchema and KeyedSerializationSchema) a utiliy to create Kafka serializers from Flink's TypeInformation system: TypeInformationKeyValueSerializationSchema.

I tried to make this change not API breaking.

[StephanEwen]

Looks quite good.

I was wondering whether the KeyedSerializationSchema should have two methods, one to serialize key, one to serialize the value. That way one does not need this Tuple2 charade, and it looks more natural to Scala users as well...

[rmetzger]

Thank you for the review.

I'll change the KeyedSerializationSchema interface.
Do you think we should merge this into 0.10.1 as well?

[StephanEwen]

It is a new feature in some sense, but it does not break anything, so probably no harm in adding it to 0.10.1 as well...

[rmetzger]

I agree. Its a new feature and under normal circumstances we would not add this to a bugfix release.
However, I would really like to merge it for 0.10.1 because a user I'm helping really needs this for using Flink.

I've changed the KeyedSerializationSchema.

[aljoscha]

Looks good, we should harmonize the generic types of KeyedSerializationSchema, SerializationSchema and DeserializationSchema, though. I.e. some have R and some don't.

[StephanEwen]

+1 from me as well

[rmetzger]

Thank you for the review. I'll merge the PR. Afterwards, I'll open a PR for harmonizing the SerializationSchema interface.