C0urante commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1266917499
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -65,7 +180,8 @@ public SinkRecord newRecord(String topic, Integer
kafkaPartition, Schema keySche
@Override
public SinkRecord newRecord(String topic, Integer kafkaPartition, Schema
keySchema, Object key, Schema valueSchema, Object value,
Long timestamp, Iterable<Header> headers) {
- return new SinkRecord(topic, kafkaPartition, keySchema, key,
valueSchema, value, kafkaOffset(), timestamp, timestampType, headers);
+ return new SinkRecord(topic, kafkaPartition, keySchema, key,
valueSchema, value, kafkaOffset(), timestamp, timestampType, headers,
+ originalTopic(), originalKafkaPartition(),
originalKafkaOffset());
Review Comment:
There's a bug in the [third
example](https://github.com/apache/kafka/blob/a1f6ab69387deb10988461152a0087f0cd2827c4/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/InternalSinkRecord.java#L54-L55)
😱
The `newRecord` method accepts an `Iterable<Header> headers` parameter, but
then when constructing the `InternalSinkRecord` that's used as the return
value, it passes in `headers()` instead of `headers` as the parameter for the
new record's headers, which causes the headers that were supplied to the
`newRecord` method to be ignored.
It'd be great if we could fix that bug (and possibly add testing coverage
for it). And IMO this is a decent reason to abandon using methods to
differentiate between fields and parameters, though as long as there aren't any
bugs in this PR I could live with it if you feel strongly that it's still worth
making that distinction.
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -65,7 +180,8 @@ public SinkRecord newRecord(String topic, Integer
kafkaPartition, Schema keySche
@Override
public SinkRecord newRecord(String topic, Integer kafkaPartition, Schema
keySchema, Object key, Schema valueSchema, Object value,
Long timestamp, Iterable<Header> headers) {
- return new SinkRecord(topic, kafkaPartition, keySchema, key,
valueSchema, value, kafkaOffset(), timestamp, timestampType, headers);
+ return new SinkRecord(topic, kafkaPartition, keySchema, key,
valueSchema, value, kafkaOffset(), timestamp, timestampType, headers,
+ originalTopic(), originalKafkaPartition(),
originalKafkaOffset());
Review Comment:
There's a bug in the [third
example](https://github.com/apache/kafka/blob/a1f6ab69387deb10988461152a0087f0cd2827c4/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/InternalSinkRecord.java#L54-L55)
😱
The `newRecord` method accepts an `Iterable<Header> headers` parameter, but
then when constructing the `InternalSinkRecord` that's used as the return
value, it passes in `headers()` instead of `headers` as the parameter for the
new record's headers, which causes the headers that were supplied to the
`newRecord` method to be ignored.
It'd be great if we could fix that bug (and possibly add testing coverage
for it). And IMO this is a decent reason to abandon using methods to
differentiate between fields and parameters, though as long as there aren't any
bugs in this PR I could live with it if you feel strongly that it's still worth
making that distinction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
