Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/2369#discussion_r75569347
--- Diff:
flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
---
@@ -207,15 +207,14 @@ public void
restoreOffsets(HashMap<KafkaTopicPartition, Long> snapshotState) {
//
------------------------------------------------------------------------
/**
- *
* <p>Implementation Note: This method is kept brief to be JIT inlining
friendly.
* That makes the fast path efficient, the extended paths are called as
separate methods.
- *
* @param record The record to emit
* @param partitionState The state of the Kafka partition from which
the record was fetched
- * @param offset The offset from which the record was fetched
+ * @param offset The offset of the record
+ * @param kafkaRecord The original Kafka record
*/
- protected final void emitRecord(T record, KafkaTopicPartitionState<KPH>
partitionState, long offset) {
+ protected <R> void emitRecord(T record, KafkaTopicPartitionState<KPH>
partitionState, long offset, R kafkaRecord) throws Exception {
--- End diff --
Is there a reason we need to have an extra `kafkaRecord`? It doesn't seem
to be used in the function.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
