Mailing-List: contact dev-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@kafka.apache.org
From: =?utf-8?Q?Michael_Andr=C3=A9_Pearce?= <michael.andre.pearce@me.com>
Content-type: text/plain; charset=utf-8
Content-transfer-encoding: quoted-printable
MIME-version: 1.0 (1.0)
Date: Wed, 13 Dec 2017 05:36:37 +0000
Subject: Re: [DISCUSS] KIP 145 - Expose Record Headers in Kafka Connect
Message-id: <A9B81EAC-2E51-48BE-9819-F14143C7A5E0@me.com>
References: <AA0D3563-37DB-4821-9280-1B53DEE1C6CB@ig.com>
 <CAA9QSJJoTvHcx+7pVH1742HF75VGcfW748JhLmGou2FjJE1bjg@mail.gmail.com>
 <87f1063ce0ef460090596e94690b06c2@BMPRDEXC142.IGI.IG.LOCAL>
 <CAE1jLMPWX1FuM2xDE9hWgW9V7OtMtVhD=S6e8wuy4H7p2wvRWg@mail.gmail.com>
 <3427DE88-0CA4-4307-BD52-D9078E99A9CF@me.com>
 <CAE1jLMOZtFgF4yC9unzges_1qu=YfiW3-RF7=1RBAcSGkom4fg@mail.gmail.com>
 <60A37B8E-F8E8-49EE-90A8-E7B5F6C7E012@ig.com>
 <fca6780b0b30425cab043e232d165107@BMPRDEXC142.IGI.IG.LOCAL>
 <65539CBD-888D-4360-AF3C-AB05079F7196@me.com>
 <CALYgK0Gqq8mjNGX_d4=eXEmCWiZ5XU7qJPmSMtQWa5GaOJyVzg@mail.gmail.com>
 <8F2867A7-D8E9-4256-B7D1-D46B91188C84@me.com>
In-reply-to: <8F2867A7-D8E9-4256-B7D1-D46B91188C84@me.com>
To: dev@kafka.apache.org
archived-at: Wed, 13 Dec 2017 05:36:56 -0000

*concert =3D convert=20

Sent from my iPhone

> On 13 Dec 2017, at 05:35, Michael Andr=C3=A9 Pearce <michael.andre.pearce@=
me.com> wrote:
>=20
> Hi Randall
>=20
> What=E2=80=99s the main difference between this and my earlier alternative=
 option PR
> https://github.com/apache/kafka/pull/2942/files
>=20
> If none then +1.
> =46rom what I can tell the only difference I make is the headers you suppo=
rt being able to cross convert primitive types eg if value after conversion i=
s integer you can still ask for float and it will type concert if possible.
>=20
> Cheers
> Mike
>=20
>=20
> Sent from my iPhone
>=20
>> On 13 Dec 2017, at 01:36, Randall Hauch <rhauch@gmail.com> wrote:
>>=20
>> Trying to revive this after several months of inactivity....
>>=20
>> I've spent quite a bit of time evaluating the current KIP-145 proposal an=
d
>> several of the suggested PRs. The original KIP-145 proposal is relatively=

>> minimalist (which is very nice), and it adopts Kafka's approach to header=
s
>> where header keys are strings and header values are byte arrays. IMO, thi=
s
>> places too much responsibility on the connector developers to know how to=

>> serialize and deserialize, which means that it's going to be difficult to=

>> assemble into pipelines connectors and stream processors that make
>> different, incompatible assumptions. It also makes Connect headers very
>> different than Connect's keys and values, which are generally structured
>> and describable with Connect schemas. I think we need Connect headers to d=
o
>> more.
>>=20
>> The other proposals attempt to do more, but even my first proposal doesn'=
t
>> seem to really provide a solution that works for Connect users and
>> connector developers. After looking at this feature from a variety of
>> perspectives over several months, I now assert that Connect must solve tw=
o
>> orthogonal problems:
>>=20
>> 1) Serialization: How different data types are (de)serialized as header
>> values
>> 2) Conversion: How values of one data type are converted to values of
>> another data type
>>=20
>> For the serialization problem, Ewen suggested quite a while back that we
>> use something akin to `Converter` for header values. Unfortunately we can=
't
>> directly reuse `Converters` since the method signatures don't allow us to=

>> supply the header name and the topic name, but we could define a
>> `HeaderConverter` that is similar to and compatible with `Converter` such=

>> that a single class could implement both. This would align Connector
>> headers with how message keys and values are handled. Each connector coul=
d
>> define which converter it wants to use; for backward compatibility purpos=
es
>> we use a header converter by default that serialize values to strings. If=

>> you want something other than this default, you'd have to specify the
>> header converter options as part of the connector configuration; this
>> proposal changes the `StringConverter`, `ByteArrayConverter`, and
>> `JsonConverter` to all implement `HeaderConverter`, so these are all
>> options. This approach supposes that a connector will serialize all of it=
s
>> headers in the same way -- with string-like representations by default. I=

>> think this is a safe assumption for the short term, and if we need more
>> control to (de)serialize named headers differently for the same connector=
,
>> we can always implement a different `HeaderConverter` that gives users mo=
re
>> control.
>>=20
>> So that would solve the serialization problem. How about connectors and
>> transforms that are implemented to expect a certain type of header value,=

>> such as an integer or boolean or timestamp? We could solve this problem
>> (for the most part) by adding methods to the `Header` interface to get th=
e
>> value in the desired type, and to support all of the sensible conversions=

>> between Connect's primitives and logical types. So, a connector or
>> transform could always call `header.valueAsObject()` to get the raw
>> representation from the converter, but a connector or transform could als=
o
>> get the string representation by calling `header.valueAsString()`, or the=

>> INT64 representation by calling `header.valueAsLong()`, etc. We could eve=
n
>> have converting methods for the built-in logical types (e.g.,
>> `header.valueAsTimestamp()` to return a java.util.Date value that is
>> described by Connect's Timestamp logical type). We can convert between mo=
st
>> primitive and logical types (e.g., anything to a STRING, INT32 to FLOAT32=
,
>> etc.), but there are a few that don't make sense (e.g., ARRAY to FLOAT32,=

>> INT32 to STRUCT, BYTE_ARRAY to anything, etc.), so these can throw a
>> `DataException`.
>>=20
>> I've refined this approach over the last few months, and have a PR for a
>> complete prototype that demonstrates these concepts and techniques:
>> https://github.com/apache/kafka/pull/4319
>>=20
>> This PR does *not* update the documentation, though I can add that if we
>> approve of this approach. And, we probably want to define (at least on th=
e
>> KIP) some relatively obvious SMTs for copying header values into record
>> key/value fields, and extracting record key/value fields into header valu=
es.
>>=20
>> @Michael, would you mind if I edited KIP-145 to reflect this proposal? I
>> would be happy to keep the existing proposal at the end of the document (=
or
>> remove it if you prefer, since it's already in the page history), and we
>> can revise as we choose a direction.
>>=20
>> Comments? Thoughts?
>>=20
>> Best regards,
>>=20
>> Randall
>>=20
>>=20
>> On Thu, Oct 19, 2017 at 2:10 PM, Michael Andr=C3=A9 Pearce <
>> michael.andre.pearce@me.com> wrote:
>>=20
>>> @rhauch
>>>=20
>>> Here is the previous discussion thread, just reigniting so we can discus=
s
>>> against the original kip thread
>>>=20
>>>=20
>>> Cheers
>>>=20
>>> Mike
>>>=20
>>> Sent from my iPhone
>>>=20
>>>> On 5 May 2017, at 02:21, Michael Pearce <Michael.Pearce@ig.com> wrote:
>>>>=20
>>>> Hi Ewen,
>>>>=20
>>>> Did you get a chance to look at the updated sample showing the idea?
>>>>=20
>>>> Did it help?
>>>>=20
>>>> Cheers
>>>> Mike
>>>>=20
>>>> Sent using OWA for iPhone
>>>> ________________________________________
>>>> From: Michael Pearce <Michael.Pearce@ig.com>
>>>> Sent: Wednesday, May 3, 2017 10:11:55 AM
>>>> To: dev@kafka.apache.org
>>>> Subject: Re: [DISCUSS] KIP 145 - Expose Record Headers in Kafka Connect=

>>>>=20
>>>> Hi Ewen,
>>>>=20
>>>> As code I think helps, as I don=E2=80=99t think I explained what I mean=
t very
>>> well.
>>>>=20
>>>> I have pushed what I was thinking to the branch/pr.
>>>> https://github.com/apache/kafka/pull/2942
>>>>=20
>>>> The key bits added on top here are:
>>>> new ConnectHeader that holds the header key (as string) and then header=

>>> value object header value schema
>>>>=20
>>>> new SubjectConverter which allows exposing a subject, in this case the
>>> subject is the key. - this can be used to register the header type in re=
pos
>>> like schema registry, or in my case below in a property file.
>>>>=20
>>>>=20
>>>> We can default the subject converter to String based of Byte based wher=
e
>>> all header values are treated safely as String or byte[] type.
>>>>=20
>>>> But this way you could add in your own converter which could be more
>>> sophisticated and convert the header based on the key.
>>>>=20
>>>> The main part is to have access to the key, so you can look up the
>>> header value type, based on the key from somewhere, aka a properties fil=
e,
>>> or some central repo (aka schema repo), where the repo subject could be t=
he
>>> topic + key, or just key if key type is global, and the schema could be
>>> primitive, String, byte[] or even can be more elaborate.
>>>>=20
>>>> Cheers
>>>> Mike
>>>>=20
>>>> On 03/05/2017, 06:00, "Ewen Cheslack-Postava" <ewen@confluent.io> wrote=
:
>>>>=20
>>>>  Michael,
>>>>=20
>>>>  Aren't JMS headers an example where the variety is a problem? Unless
>>> I'm
>>>>  misunderstanding, there's not even a fixed serialization format
>>> expected
>>>>  for them since JMS defines the runtime types, not the wire format. For=

>>>>  example, we have JMSCorrelationID (String), JMSExpires (Long), and
>>>>  JMSReplyTo (Destination). These are simply run time types, so we'd
>>> need
>>>>  either (a) a different serializer/deserializer for each or (b) a
>>>>  serializer/deserializer that can handle all of them (e.g. Avro, JSON,
>>> etc).
>>>>=20
>>>>  What is the actual serialized format of the different fields? And if
>>> it's
>>>>  not specified anywhere in the KIP, why should using the well-known
>>> type for
>>>>  the header key (e.g. use StringSerializer, IntSerializer, etc) be
>>> better or
>>>>  worse than using a general serialization format (e.g. Avro, JSON)?
>>> And if
>>>>  the latter is the choice, how do you decide on the format?
>>>>=20
>>>>  -Ewen
>>>>=20
>>>>  On Tue, May 2, 2017 at 12:48 PM, Michael Andr=C3=A9 Pearce <
>>>>  michael.andre.pearce@me.com> wrote:
>>>>=20
>>>>> Hi Ewan,
>>>>>=20
>>>>> So on the point of JMS the predefined/standardised JMS and JMSX header=
s
>>>>> have predefined types. So these can be serialised/deserialised
>>> accordingly.
>>>>>=20
>>>>> Custom jms headers agreed could be a bit more difficult but on the 80/=
20
>>>>> rule I would agree mostly they're string values and as anyhow you can
>>> hold
>>>>> bytes as a string it wouldn't cause any issue, defaulting to that.
>>>>>=20
>>>>> But I think easily we maybe able to do one better.
>>>>>=20
>>>>> Obviously can override the/config the headers converter but we can
>>> supply
>>>>> a default converter could take a config file with key to type mapping?=

>>>>>=20
>>>>> Allowing people to maybe define/declare a header key with the expected=

>>>>> type in some property file? To support string, byte[] and primitives?
>>> And
>>>>> undefined headers just either default to String or byte[]
>>>>>=20
>>>>> We could also pre define known headers like the jms ones mentioned
>>> above.
>>>>>=20
>>>>> E.g
>>>>>=20
>>>>> AwesomeHeader1=3Dboolean
>>>>> AwesomeHeader2=3Dlong
>>>>> JMSCorrelationId=3DString
>>>>> JMSXGroupId=3DString
>>>>>=20
>>>>>=20
>>>>> What you think?
>>>>>=20
>>>>>=20
>>>>> Cheers
>>>>> Mike
>>>>>=20
>>>>>=20
>>>>>=20
>>>>>=20
>>>>>=20
>>>>>=20
>>>>> Sent from my iPhone
>>>>>=20
>>>>>> On 2 May 2017, at 18:45, Ewen Cheslack-Postava <ewen@confluent.io>
>>>>> wrote:
>>>>>>=20
>>>>>> A couple of thoughts:
>>>>>>=20
>>>>>> First, agreed that we definitely want to expose header functionality.=

>>>>> Thank
>>>>>> you Mike for starting the conversation! Even if Connect doesn't do
>>>>> anything
>>>>>> special with it, there's value in being able to access/set headers.
>>>>>>=20
>>>>>> On motivation -- I think there are much broader use cases. When
>>> thinking
>>>>>> about exposing headers, I'd actually use Replicator as only a minor
>>>>>> supporting case. The reason is that it is a very uncommon case where
>>>>> there
>>>>>> is zero impedance mismatch between the source and sink of the data
>>> since
>>>>>> they are both Kafka. This means you don't need to think much about da=
ta
>>>>>> formats/serialization. I think the JMS use case is a better example
>>> since
>>>>>> JMS headers and Kafka headers don't quite match up. Here's a quick li=
st
>>>>> of
>>>>>> use cases I can think of off the top of my head:
>>>>>>=20
>>>>>> 1. Include headers from other systems that support them: JMS (or real=
ly
>>>>> any
>>>>>> MQ), HTTP
>>>>>> 2. Other connector-specific headers. For example, from JDBC maybe the=

>>>>> table
>>>>>> the data comes from is a header; for a CDC connector you might includ=
e
>>>>> the
>>>>>> binlog offset as a header.
>>>>>> 3. Interceptor/SMT-style use cases for annotating things like
>>> provenance
>>>>> of
>>>>>> data:
>>>>>> 3a. Generically w/ user-supplied data like data center, host, app ID,=

>>>>> etc.
>>>>>> 3b. Kafka Connect framework level info, such as the connector/task
>>>>>> generating the data
>>>>>>=20
>>>>>> On deviation from Connect's model -- to be honest, the KIP-82 also
>>>>> deviates
>>>>>> quite substantially from how Kafka handles data already, so we may
>>>>> struggle
>>>>>> a bit to rectify the two. (In particular, headers specify some
>>> structure
>>>>>> and enforce strings specifically for header keys, but then require yo=
u
>>> to
>>>>>> do serialization of header values yourself...).
>>>>>>=20
>>>>>> I think the use cases I mentioned above may also need different
>>>>> approaches
>>>>>> to how the data in headers are handled. As Gwen mentions, if we expos=
e
>>>>> the
>>>>>> headers to Connectors, they need to have some idea of the format and
>>> the
>>>>>> reason for byte[] values in KIP-82 is to leave that decision up to th=
e
>>>>>> organization using them. But without knowing the format, connectors
>>> can't
>>>>>> really do anything with them -- if a source connector assumes a forma=
t,
>>>>>> they may generate data incompatible with the format used by the rest o=
f
>>>>> the
>>>>>> organization. On the other hand, I have a feeling most people will ju=
st
>>>>> use
>>>>>> <String, String> headers, so allowing connectors to embed arbitrarily=

>>>>>> complex data may not work out well in practice. Or maybe we leave it
>>>>>> flexible, most people default to using StringConverter for the
>>> serializer
>>>>>> and Connectors will end up defaulting to that just for compatibility.=
..
>>>>>>=20
>>>>>> I'm not sure I have a real proposal yet, but I do think understanding=

>>> the
>>>>>> impact of using a Converter for headers would be useful, and we might=

>>>>> want
>>>>>> to think about how this KIP would fit in with transformations (or if
>>> that
>>>>>> is something that can be deferred, handled separately from the existi=
ng
>>>>>> transformations, etc).
>>>>>>=20
>>>>>> -Ewen
>>>>>>=20
>>>>>> On Mon, May 1, 2017 at 11:52 AM, Michael Pearce <Michael.Pearce@ig.co=
m
>>>>=20
>>>>>> wrote:
>>>>>>=20
>>>>>>> Hi Gwen,
>>>>>>>=20
>>>>>>> Then intent here was to allow tools that perform similar role to
>>> mirror
>>>>>>> makers of replicating the messaging from one cluster to another.  Eg=

>>>>> like
>>>>>>> mirror make should just be taking and transferring the headers as is=
.
>>>>>>>=20
>>>>>>> We don't actually use this inside our company, so not exposing this
>>>>> isn't
>>>>>>> an issue for us. Just believe there are companies like confluent who=

>>>>> have
>>>>>>> tools like replicator that do.
>>>>>>>=20
>>>>>>> And as good citizens think we should complete the work and expose th=
e
>>>>>>> headers same as in the record to at least allow them to replicate th=
e
>>>>>>> messages as is. Note Steph seems to want it.
>>>>>>>=20
>>>>>>> Cheers
>>>>>>> Mike
>>>>>>>=20
>>>>>>> Sent using OWA for iPhone
>>>>>>> ________________________________________
>>>>>>> From: Gwen Shapira <gwen@confluent.io>
>>>>>>> Sent: Monday, May 1, 2017 2:36:34 PM
>>>>>>> To: dev@kafka.apache.org
>>>>>>> Subject: Re: [DISCUSS] KIP 145 - Expose Record Headers in Kafka
>>> Connect
>>>>>>>=20
>>>>>>> Hi,
>>>>>>>=20
>>>>>>> I'm excited to see the community expanding Connect in this direction=
!
>>>>>>> Headers + Transforms =3D=3D Fun message routing.
>>>>>>>=20
>>>>>>> I like how clean the proposal is, but I'm concerned that it kinda
>>>>> deviates
>>>>>>> from how Connect handles data elsewhere.
>>>>>>> Unlike Kafka, Connect doesn't look at all data as byte-arrays, we ha=
ve
>>>>>>> converters that take data in specific formats (JSON, Avro) and turns=

>>> it
>>>>>>> into Connect data types (defined in the data api). I think it will b=
e
>>>>> more
>>>>>>> consistent for connector developers to also get headers as some kind=

>>> of
>>>>>>> structured or semi-structured data (and to expand the converters to
>>>>> handle
>>>>>>> header conversions as well).
>>>>>>> This will allow for Connect's separation of concerns - Connector
>>>>> developers
>>>>>>> don't worry about data formats (because they get the internal connec=
t
>>>>>>> objects) and Converters do all the data format work.
>>>>>>>=20
>>>>>>> Another thing, in my experience, APIs work better if they are put in=
to
>>>>> use
>>>>>>> almost immediately - so difficulties in using the APIs are immediate=
ly
>>>>>>> surfaced. Are you planning any connectors that will use this feature=

>>>>> (not
>>>>>>> necessarily in Kafka, just in general)? Or perhaps we can think of a=

>>>>> way to
>>>>>>> expand Kafka's file connectors so they'll use headers somehow (can't=

>>>>> think
>>>>>>> of anything, but maybe?).
>>>>>>>=20
>>>>>>> Gwen
>>>>>>>=20
>>>>>>> On Sat, Apr 29, 2017 at 12:12 AM, Michael Pearce <
>>> Michael.Pearce@ig.com
>>>>>>=20
>>>>>>> wrote:
>>>>>>>=20
>>>>>>>> Hi All,
>>>>>>>>=20
>>>>>>>> Now KIP-82 is committed I would like to discuss extending the work t=
o
>>>>>>>> expose it in Kafka Connect, its primary focus being so connectors
>>> that
>>>>>>> may
>>>>>>>> do similar tasks as MirrorMakers, either Kafka->Kafka or JMS-Kafka
>>>>> would
>>>>>>> be
>>>>>>>> able to replicate the headers.
>>>>>>>> It would be ideal but not mandatory for this to go in 0.11 release s=
o
>>>>> is
>>>>>>>> available on day one of headers being available.
>>>>>>>>=20
>>>>>>>> Please find the KIP here:
>>>>>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
>>>>>>>> 145+-+Expose+Record+Headers+in+Kafka+Connect
>>>>>>>>=20
>>>>>>>> Please find an initial implementation as a PR here:
>>>>>>>> https://github.com/apache/kafka/pull/2942
>>>>>>>>=20
>>>>>>>> Kind Regards
>>>>>>>> Mike
>>>>>>>> The information contained in this email is strictly confidential an=
d
>>>>> for
>>>>>>>> the use of the addressee only, unless otherwise indicated. If you a=
re
>>>>> not
>>>>>>>> the intended recipient, please do not read, copy, use or disclose t=
o
>>>>>>> others
>>>>>>>> this message or any attachment. Please also notify the sender by
>>>>> replying
>>>>>>>> to this email or by telephone (+44(020 7896 0011) and then delete t=
he
>>>>>>> email
>>>>>>>> and any copies of it. Opinions, conclusion (etc) that do not relate=

>>> to
>>>>>>> the
>>>>>>>> official business of this company shall be understood as neither
>>> given
>>>>>>> nor
>>>>>>>> endorsed by it. IG is a trading name of IG Markets Limited (a compa=
ny
>>>>>>>> registered in England and Wales, company number 04008957) and IG
>>> Index
>>>>>>>> Limited (a company registered in England and Wales, company number
>>>>>>>> 01190902). Registered address at Cannon Bridge House, 25 Dowgate
>>> Hill,
>>>>>>>> London EC4R 2YA. Both IG Markets Limited (register number 195355) a=
nd
>>>>> IG
>>>>>>>> Index Limited (register number 114059) are authorised and regulated=

>>> by
>>>>>>> the
>>>>>>>> Financial Conduct Authority.
>>>>>>>>=20
>>>>>>>=20
>>>>>>>=20
>>>>>>>=20
>>>>>>> --
>>>>>>> *Gwen Shapira*
>>>>>>> Product Manager | Confluent
>>>>>>> 650.450.2760 | @gwenshap
>>>>>>> Follow us: Twitter <https://twitter.com/ConfluentInc> | blog
>>>>>>> <http://www.confluent.io/blog>
>>>>>>> The information contained in this email is strictly confidential and=

>>> for
>>>>>>> the use of the addressee only, unless otherwise indicated. If you ar=
e
>>>>> not
>>>>>>> the intended recipient, please do not read, copy, use or disclose to=

>>>>> others
>>>>>>> this message or any attachment. Please also notify the sender by
>>>>> replying
>>>>>>> to this email or by telephone (+44(020 7896 0011) and then delete th=
e
>>>>> email
>>>>>>> and any copies of it. Opinions, conclusion (etc) that do not relate t=
o
>>>>> the
>>>>>>> official business of this company shall be understood as neither giv=
en
>>>>> nor
>>>>>>> endorsed by it. IG is a trading name of IG Markets Limited (a compan=
y
>>>>>>> registered in England and Wales, company number 04008957) and IG Ind=
ex
>>>>>>> Limited (a company registered in England and Wales, company number
>>>>>>> 01190902). Registered address at Cannon Bridge House, 25 Dowgate Hil=
l,
>>>>>>> London EC4R 2YA. Both IG Markets Limited (register number 195355) an=
d
>>> IG
>>>>>>> Index Limited (register number 114059) are authorised and regulated b=
y
>>>>> the
>>>>>>> Financial Conduct Authority.
>>>>>>>=20
>>>>>=20
>>>>=20
>>>>=20
>>>> The information contained in this email is strictly confidential and fo=
r
>>> the use of the addressee only, unless otherwise indicated. If you are no=
t
>>> the intended recipient, please do not read, copy, use or disclose to oth=
ers
>>> this message or any attachment. Please also notify the sender by replyin=
g
>>> to this email or by telephone (+44(020 7896 0011) and then delete the em=
ail
>>> and any copies of it. Opinions, conclusion (etc) that do not relate to t=
he
>>> official business of this company shall be understood as neither given n=
or
>>> endorsed by it. IG is a trading name of IG Markets Limited (a company
>>> registered in England and Wales, company number 04008957) and IG Index
>>> Limited (a company registered in England and Wales, company number
>>> 01190902). Registered address at Cannon Bridge House, 25 Dowgate Hill,
>>> London EC4R 2YA. Both IG Markets Limited (register number 195355) and IG=

>>> Index Limited (register number 114059) are authorised and regulated by t=
he
>>> Financial Conduct Authority.
>>>=20