{"action":"created","issue":{"url":"https://api.github.com/repos/apache/flink/issues/6615","repository_url":"https://api.github.com/repos/apache/flink","labels_url":"https://api.github.com/repos/apache/flink/issues/6615/labels{/name}","comments_url":"https://api.github.com/repos/apache/flink/issues/6615/comments","events_url":"https://api.github.com/repos/apache/flink/issues/6615/events","html_url":"https://github.com/apache/flink/pull/6615","id":354126421,"node_id":"MDExOlB1bGxSZXF1ZXN0MjEwOTY5Mzk5","number":6615,"title":"[FLINK-8354] [flink-connectors] Add ability to access and provider Kafka headers","user":{"login":"alexeyt820","id":12479962,"node_id":"MDQ6VXNlcjEyNDc5OTYy","avatar_url":"https://avatars0.githubusercontent.com/u/12479962?v=4","gravatar_id":"","url":"https://api.github.com/users/alexeyt820","html_url":"https://github.com/alexeyt820","followers_url":"https://api.github.com/users/alexeyt820/followers","following_url":"https://api.github.com/users/alexeyt820/following{/other_user}","gists_url":"https://api.github.com/users/alexeyt820/gists{/gist_id}","starred_url":"https://api.github.com/users/alexeyt820/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alexeyt820/subscriptions","organizations_url":"https://api.github.com/users/alexeyt820/orgs","repos_url":"https://api.github.com/users/alexeyt820/repos","events_url":"https://api.github.com/users/alexeyt820/events{/privacy}","received_events_url":"https://api.github.com/users/alexeyt820/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":6,"created_at":"2018-08-26T20:31:36Z","updated_at":"2018-10-22T14:58:14Z","closed_at":null,"author_association":"NONE","pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/6615","html_url":"https://github.com/apache/flink/pull/6615","diff_url":"https://github.com/apache/flink/pull/6615.diff","patch_url":"https://github.com/apache/flink/pull/6615.patch"},"body":"\r\n## What is the purpose of the change\r\nThis changes allows to access Kafka headers in KeyedDeserializationSchema, and to provide Kafka headers via KeyedSerializationSchema.headers method\r\n\r\n## Brief change log\r\n- Add default _headers_ method to KeyedSerializationSchema interface\r\n- Add default deserialize method with _headers_ argument to KeyedDeserializationSchema\r\n- Add Kafka011Fetcher to get headers from Kafka ConsumerRecord and provide them as parameter KeyedDeserializationSchema\r\n- Modify FlinkKafkaProducer011 to request _headers_ from KeyedSerializationSchema and add them to ProducerRecord.\r\n\r\n## Verifying this change\r\n\r\nThis change added tests and can be verified as follows:\r\n  - *Added integration test *Kafka011ITCase.testHeaders* to verify that we can produce headers and consume them\r\n\r\n## Does this pull request potentially affect one of the following parts:\r\n\r\n  - Dependencies (does it add or upgrade a dependency): (no)\r\n  - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes)\r\n  - The serializers: (no)\r\n  - The runtime per-record code paths (performance sensitive): (yes)\r\n  - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)\r\n  - The S3 file system connector: (no)\r\n\r\n## Documentation\r\n\r\n  - Does this pull request introduce a new feature? (yes)\r\n  - If yes, how is the feature documented? (JavaDocs)"},"comment":{"url":"https://api.github.com/repos/apache/flink/issues/comments/431860212","html_url":"https://github.com/apache/flink/pull/6615#issuecomment-431860212","issue_url":"https://api.github.com/repos/apache/flink/issues/6615","id":431860212,"node_id":"MDEyOklzc3VlQ29tbWVudDQzMTg2MDIxMg==","user":{"login":"alexeyt820","id":12479962,"node_id":"MDQ6VXNlcjEyNDc5OTYy","avatar_url":"https://avatars0.githubusercontent.com/u/12479962?v=4","gravatar_id":"","url":"https://api.github.com/users/alexeyt820","html_url":"https://github.com/alexeyt820","followers_url":"https://api.github.com/users/alexeyt820/followers","following_url":"https://api.github.com/users/alexeyt820/following{/other_user}","gists_url":"https://api.github.com/users/alexeyt820/gists{/gist_id}","starred_url":"https://api.github.com/users/alexeyt820/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alexeyt820/subscriptions","organizations_url":"https://api.github.com/users/alexeyt820/orgs","repos_url":"https://api.github.com/users/alexeyt820/repos","events_url":"https://api.github.com/users/alexeyt820/events{/privacy}","received_events_url":"https://api.github.com/users/alexeyt820/received_events","type":"User","site_admin":false},"created_at":"2018-10-22T14:58:14Z","updated_at":"2018-10-22T14:58:14Z","author_association":"NONE","body":"Rebased on top of #6703 and implemented KeyedDeserializationSchema.Record for Kafka 2.0 (had to keep KeyedDeserializationSchema.Record because KeyedDeserializationSchema is used by legacy connectors as well"}}
{"actor":{"display_login":"alexeyt820"}}