You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "yashmayya (via GitHub)" <gi...@apache.org> on 2023/07/17 06:55:59 UTC
[GitHub] [kafka] yashmayya opened a new pull request, #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
yashmayya opened a new pull request, #14024:
URL: https://github.com/apache/kafka/pull/14024
- https://cwiki.apache.org/confluence/display/KAFKA/KIP-793%3A+Allow+sink+connectors+to+be+used+with+topic-mutating+SMTs
- https://issues.apache.org/jira/browse/KAFKA-13431
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] gharris1727 commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1265731323
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * name when deployed to older Connect runtimes.
+ *
+ * @return the topic corresponding to the Kafka record before any transformations were applied
+ *
+ * @since 3.6
+ */
+ public String originalTopic() {
+ return originalTopic;
+ }
+
+ /**
+ * Get the original topic partition for this sink record, corresponding to the topic partition of the Kafka record
+ * before any {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should
+ * be used by sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #kafkaPartition()}, in order to be compatible
+ * with transformations that mutate the topic partition value.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalKafkaPartition;
+ * try {
+ * originalKafkaPartition = record.originalKafkaPartition();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalKafkaPartition = record.kafkaPartition();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * partition when deployed to older Connect runtimes.
+ *
+ * @return the topic partition corresponding to the Kafka record before any transformations were applied
+ *
+ * @since 3.6
+ */
+ public Integer originalKafkaPartition() {
+ return originalKafkaPartition;
+ }
+
+ /**
+ * Get the original offset for this sink record, corresponding to the offset of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #kafkaOffset()}, in order to be
+ * compatible with transformations that mutate the offset value.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalKafkaOffset;
+ * try {
+ * originalKafkaOffset = record.originalKafkaOffset();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
Review Comment:
I think the NoClassDefFoundError was specific to the error reporter, since the method returned a type which didn't always exist. Here, where the return types are native Java types, I think the NoSuchMethodError is the only error possible here. Can you verify this?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/InternalSinkRecord.java:
##########
@@ -34,27 +31,12 @@ public class InternalSinkRecord extends SinkRecord {
public InternalSinkRecord(ConsumerRecord<byte[], byte[]> originalRecord, SinkRecord record) {
super(record.topic(), record.kafkaPartition(), record.keySchema(), record.key(),
- record.valueSchema(), record.value(), record.kafkaOffset(), record.timestamp(),
- record.timestampType(), record.headers());
+ record.valueSchema(), record.value(), record.kafkaOffset(), record.timestamp(),
+ record.timestampType(), record.headers(), originalRecord.topic(), originalRecord.partition(),
+ originalRecord.offset());
this.originalRecord = originalRecord;
}
- protected InternalSinkRecord(ConsumerRecord<byte[], byte[]> originalRecord, String topic,
- int partition, Schema keySchema, Object key, Schema valueSchema,
- Object value, long kafkaOffset, Long timestamp,
- TimestampType timestampType, Iterable<Header> headers) {
- super(topic, partition, keySchema, key, valueSchema, value, kafkaOffset, timestamp, timestampType, headers);
- this.originalRecord = originalRecord;
- }
-
- @Override
- public SinkRecord newRecord(String topic, Integer kafkaPartition, Schema keySchema, Object key,
Review Comment:
This override was added due to the following comment: https://github.com/apache/kafka/pull/8720/files#r431367467
But I don't understand the necessity of having the override. Is there a reason you needed to remove this override? Without understanding the need to include or exclude it, i'd err on keeping what was already there.
As far as I can tell, the InternalSinkRecord is only passed to the connector. I am not aware of any use-cases for calling newRecord in the connector, but it is of course possible.
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
Review Comment:
The connector can receive two SinkRecords which may pass equality if they have the same T/P/O, and the same contents, header, etc. Right now this can happen when an SMT causes a T/P/O collision, and if a record is re-delivered to the connector. After this change, only a re-delivery can cause this equality check to fire.
I don't think it would be reasonable for someone to rely on this to detect T/P/O collisions, because it is inherently unreliable due to checking equality of the contents of the record. I think the semantics for only being equal after a re-delivery is much better.
We don't seem to rely on this equals method in the runtime, so this should only affect connector implementations.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante merged pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante merged PR #14024:
URL: https://github.com/apache/kafka/pull/14024
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1266903930
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
Review Comment:
👍 LGTM, thanks!
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
Review Comment:
👍 LGTM, thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #14024:
URL: https://github.com/apache/kafka/pull/14024#issuecomment-1644061803
Thanks Chris, I've rebased this on the latest `trunk`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1266413275
##########
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:
I figured this was intentional to differentiate between the use of instance fields and method arguments and chose to stick with the convention used here. Examples - [1](https://github.com/apache/kafka/blob/a1f6ab69387deb10988461152a0087f0cd2827c4/connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java#L62), [2](https://github.com/apache/kafka/blob/a1f6ab69387deb10988461152a0087f0cd2827c4/connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java#L68), [3](https://github.com/apache/kafka/blob/a1f6ab69387deb10988461152a0087f0cd2827c4/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/InternalSinkRecord.java#L54-L55)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #14024:
URL: https://github.com/apache/kafka/pull/14024#issuecomment-1645809586
Jenkins build is pretty unstable but this appears due to unrelated issues. We have a successful build (minus a few flaky tests) on two out of four nodes in the latest run. Merging...
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #14024:
URL: https://github.com/apache/kafka/pull/14024#issuecomment-1642257814
@gharris1727 did you want to give this another pass before we merge? No worries if you don't have the bandwidth.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1265675099
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
Review Comment:
On top of the `Transformation` link nit, we can simplify some of the language here:
```suggestion
* Get the original topic for this sink record, before any
* {@link Transformation transformations} were applied.
* In order to be compatible with transformations that mutate topic names, this method should be used
* by sink tasks instead of {@link #topic()} for any internal offset tracking purposes (for instance, reporting offsets to the Connect runtime via
* {@link SinkTask#preCommit(Map)}).
```
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -20,18 +20,25 @@
import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.transforms.Transformation;
+
+import java.util.Map;
+import java.util.Objects;
/**
- * SinkRecord is a {@link ConnectRecord} that has been read from Kafka and includes the kafkaOffset of
- * the record in the Kafka topic-partition in addition to the standard fields. This information
- * should be used by the {@link SinkTask} to coordinate kafkaOffset commits.
+ * SinkRecord is a {@link ConnectRecord} that has been read from Kafka and includes the original Kafka record's
+ * topic, partition and offset (before any {@link Transformation}s have been applied) in addition to the standard fields.
+ * This information should be used by the {@link SinkTask} to coordinate offset commits.
Review Comment:
Nit: formatting/readability
```suggestion
* topic, partition and offset (before any {@link Transformation transformations}
* have been applied) in addition to the standard fields. This information should be used by the {@link SinkTask} to coordinate
* offset commits.
```
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
Review Comment:
Nit: formatting/readability
```suggestion
* {@link org.apache.kafka.connect.transforms.Transformation transformations} were applied. This should be used by
```
(this applies to a bunch of places in this PR; only noting it here to reduce noise)
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * name when deployed to older Connect runtimes.
Review Comment:
Nit: language tweak, clarification on unsupported Connect runtimes
```suggestion
* Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate topic
* names when deployed to older Connect runtimes that do not support this method.
```
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * name when deployed to older Connect runtimes.
+ *
+ * @return the topic corresponding to the Kafka record before any transformations were applied
Review Comment:
Nit: can simplify language (don't love "corresponding" if we can find something less wordy)
```suggestion
* @return the topic for this record before any transformations were applied
```
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
Review Comment:
I'm torn. It doesn't seem especially safe to provide this as an example of how to handle older Connect runtimes. But it also seems a little ham-fisted to recommend that developers throw an exception with a message like "This connector is incompatible with this version of Kafka Connect".
What do you think about adding a warning log message to the `catch` body stating that, with the current version of the Connect runtime, this connector will be incompatible with TPO-mutating SMTs?
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * name when deployed to older Connect runtimes.
+ *
+ * @return the topic corresponding to the Kafka record before any transformations were applied
+ *
+ * @since 3.6
+ */
+ public String originalTopic() {
+ return originalTopic;
+ }
+
+ /**
+ * Get the original topic partition for this sink record, corresponding to the topic partition of the Kafka record
+ * before any {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should
+ * be used by sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #kafkaPartition()}, in order to be compatible
+ * with transformations that mutate the topic partition value.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalKafkaPartition;
+ * try {
+ * originalKafkaPartition = record.originalKafkaPartition();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalKafkaPartition = record.kafkaPartition();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * partition when deployed to older Connect runtimes.
+ *
+ * @return the topic partition corresponding to the Kafka record before any transformations were applied
+ *
+ * @since 3.6
+ */
+ public Integer originalKafkaPartition() {
+ return originalKafkaPartition;
+ }
+
+ /**
+ * Get the original offset for this sink record, corresponding to the offset of the Kafka record before any
Review Comment:
I think the remarks for `originalTopic` can all be applied here as well; not copying them over to reduce noise.
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * name when deployed to older Connect runtimes.
+ *
+ * @return the topic corresponding to the Kafka record before any transformations were applied
+ *
+ * @since 3.6
+ */
+ public String originalTopic() {
+ return originalTopic;
+ }
+
+ /**
+ * Get the original topic partition for this sink record, corresponding to the topic partition of the Kafka record
Review Comment:
I think the remarks for `originalTopic` can all be applied here as well; not copying them over to reduce noise.
##########
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:
Nit: for consistency's sake, can we change this to use fields directly instead of access method?
```suggestion
return new SinkRecord(topic, kafkaPartition, keySchema, key, valueSchema, value, kafkaOffset, timestamp, timestampType, headers,
originalTopic, originalKafkaPartition, originalKafkaOffset);
```
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
Review Comment:
Nit: consistent formatting
```suggestion
Objects.equals(originalKafkaPartition, that.originalKafkaPartition) &&
originalKafkaOffset == that.originalKafkaOffset;
```
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkTask.java:
##########
@@ -118,11 +122,16 @@ public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
* The default implementation simply invokes {@link #flush(Map)} and is thus able to assume all {@code currentOffsets}
* are safe to commit.
*
- * @param currentOffsets the current offset state as of the last call to {@link #put(Collection)}},
- * provided for convenience but could also be determined by tracking all offsets included in the {@link SinkRecord}s
- * passed to {@link #put}.
+ * @param currentOffsets the current offset state as of the last call to {@link #put(Collection)}, provided for
+ * convenience but could also be determined by tracking all offsets included in the
+ * {@link SinkRecord}s passed to {@link #put}. Note that the topic, partition and offset
+ * here correspond to the original Kafka topic partition and offset, before any {@link Transformation}s
+ * have been applied. These can be tracked by the task through the {@link SinkRecord#originalTopic()},
+ * {@link SinkRecord#originalKafkaPartition()} and {@link SinkRecord#originalKafkaOffset()} methods.
*
- * @return an empty map if Connect-managed offset commit is not desired, otherwise a map of offsets by topic-partition that are safe to commit.
+ * @return an empty map if Connect-managed offset commit is not desired, otherwise a map of offsets by topic-partition that are
+ * safe to commit. Note that the returned topic-partition to offsets map should also use the original Kafka
Review Comment:
Nit:
```suggestion
* safe to commit. Note that the returned topic-partition to offsets map should use the original Kafka
```
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -44,9 +51,18 @@ public SinkRecord(String topic, int partition, Schema keySchema, Object key, Sch
public SinkRecord(String topic, int partition, Schema keySchema, Object key, Schema valueSchema, Object value, long kafkaOffset,
Long timestamp, TimestampType timestampType, Iterable<Header> headers) {
+ this(topic, partition, keySchema, key, valueSchema, value, kafkaOffset, timestamp, timestampType, headers, topic, partition, kafkaOffset);
+ }
+
+ public SinkRecord(String topic, int partition, Schema keySchema, Object key, Schema valueSchema, Object value, long kafkaOffset,
Review Comment:
Can we add a Javadoc to this constructor stating that it's intended for use by the Connect runtime only, and that developers should not use it directly in plugin code?
(I'm qualifying things that way because it may still be useful for developers to use this constructor when writing unit tests for their connector/transformation/etc.)
##########
connect/api/src/test/java/org/apache/kafka/connect/sink/SinkRecordTest.java:
##########
@@ -125,4 +125,19 @@ public void shouldModifyRecordHeader() {
Header header = record.headers().lastWithName("intHeader");
assertEquals(100, (int) Values.convertToInteger(header.schema(), header.value()));
}
-}
\ No newline at end of file
+
+ @Test
+ public void shouldRetainOriginalTopicPartition() {
Review Comment:
Can we update the declaration for `record` in the `beforeEach` method to explicitly use the constructor that takes in the original TPO? Right now we're implicitly relying on that behavior with this test.
If we want to be paranoid (we don't have to but it's up to you), we can also add a test case for the older constructors that verifies that they provide the possibly-transformed TPO as the original TPO as well.
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
Review Comment:
Thanks for the detailed analysis!
Anticipating some possible use cases that may be impacted by this, I've come up with:
1. Equality testing in unit tests
2. Storing sink records in data structures (e.g., elements in a `HashSet` or keys in a `HashMap`) which contain elements that are unique according to their `equals` method
With 1, this change is unlikely to be controversial.
With 2, things are a little trickier, but ultimately I think it's best to proceed with this change. There's enough granularity in the existing `equals` method that I can't anticipate any realistic cases where that method would previously return `true` but would now return `false`. For example, if a connector is tracking records that were redelivered to it via `Sink::put` after throwing an exception in `SinkTask::preCommit`, this change won't affect that case since the re-delivered records would have the same original TPO.
Also, FWIW, there's precedent here with when we added headers to Connect in [KIP-145](https://cwiki.apache.org/confluence/display/KAFKA/KIP-145+-+Expose+Record+Headers+in+Kafka+Connect); there was no discussion on [the PR](https://github.com/apache/kafka/pull/4319) regarding the impact that changes to these methods may have on compatibility.
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
}
@Override
public int hashCode() {
int result = super.hashCode();
result = 31 * result + Long.hashCode(kafkaOffset);
result = 31 * result + timestampType.hashCode();
+ result = 31 * result + originalTopic.hashCode();
+ result = 31 * result + originalKafkaPartition.hashCode();
+ result = 31 * result + Long.hashCode(originalKafkaOffset);
Review Comment:
LGTM 👍
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
}
@Override
public int hashCode() {
int result = super.hashCode();
result = 31 * result + Long.hashCode(kafkaOffset);
result = 31 * result + timestampType.hashCode();
+ result = 31 * result + originalTopic.hashCode();
+ result = 31 * result + originalKafkaPartition.hashCode();
+ result = 31 * result + Long.hashCode(originalKafkaOffset);
Review Comment:
LGTM 👍
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #14024:
URL: https://github.com/apache/kafka/pull/14024#issuecomment-1642498648
Jenkins is failing because it's merging a commit on trunk that doesn't include https://github.com/apache/kafka/pull/14037 before testing, which causes the `rat` task for Gradle to fail.
I've pushed an empty commit to retrigger CI; hopefully that'll do the trick and pick up the latest changes from trunk.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #14024:
URL: https://github.com/apache/kafka/pull/14024#issuecomment-1644022144
@yashmayya Mind patching the merge conflicts introduced by https://github.com/apache/kafka/pull/14044? Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1267245540
##########
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:
A separate ticket/PR sounds great, thanks! Feel free to ping me on the PR if you implement the fx.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1264950055
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
Review Comment:
This change is binary compatible. Also, it shouldn't affect any existing sink connectors in any way (i.e. it is effectively source / API compatible) since the existing public constructors for `SinkRecord` have been modified to simply use `topic`, `kafkaPartition`, and `kafkaOffset` for `originalTopic`, `originalKafkaPartition`, and `originalKafkaOffset` respectively (so records that were equal earlier will continue being equal and vice versa).
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
}
@Override
public int hashCode() {
int result = super.hashCode();
result = 31 * result + Long.hashCode(kafkaOffset);
result = 31 * result + timestampType.hashCode();
+ result = 31 * result + originalTopic.hashCode();
+ result = 31 * result + originalKafkaPartition.hashCode();
+ result = 31 * result + Long.hashCode(originalKafkaOffset);
Review Comment:
Similar to the above change, this is binary compatible too. Although this one could potentially break some oddball downstream use cases like tests asserting exact hash code values for sink records. While this isn't strictly source / API compatible, it should be fairly unlikely that such use cases exist in the wild. If this is still deemed to be a concern (or more cases crop up), this change can be removed without affecting the rest of this PR / KIP.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1264950055
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -79,17 +195,21 @@ public boolean equals(Object o) {
SinkRecord that = (SinkRecord) o;
- if (kafkaOffset != that.kafkaOffset)
- return false;
-
- return timestampType == that.timestampType;
+ return kafkaOffset == that.kafkaOffset &&
+ timestampType == that.timestampType &&
+ Objects.equals(originalTopic, that.originalTopic) &&
+ Objects.equals(originalKafkaPartition, that.originalKafkaPartition)
+ && originalKafkaOffset == that.originalKafkaOffset;
Review Comment:
This change is binary compatible. Source / API compatibility is a bit trickier here however. The existing public constructors for `SinkRecord` have been modified to simply use `topic`, `kafkaPartition`, and `kafkaOffset` for `originalTopic`, `originalKafkaPartition`, and `originalKafkaOffset` respectively (so records created outside the Connect framework that were equal earlier will continue being equal and vice versa). The new constructor which includes the original topic, partition, and offset is now being used by the framework. This means that if there are 2 records with the same post-transform topic, partition, and offset but different pre-transform topic, partition or offset they will now be considered as unequal.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1266957728
##########
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:
Haha wow I completely glossed over that, nice catch.
> It'd be great if we could fix that bug (and possibly add testing coverage for it)
I think it might be worth filing a separate bug ticket and PR to fix that so that we could also backport it as well? We did discuss whether or not the method is even useful [here](https://github.com/apache/kafka/pull/14024#discussion_r1265694650), but given that we technically do support such use cases (and can't prove that it isn't used), it'd probably be good to backport a fix for such a bad bug if possible. WDYT?
> And IMO this is a decent reason to abandon using methods to differentiate between fields and parameters
Yeah, I'm definitely convinced since I missed it as well 😆
##########
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:
Haha wow I completely glossed over that, nice catch.
> It'd be great if we could fix that bug (and possibly add testing coverage for it)
I think it might be worth filing a separate bug ticket and PR to fix that so that we could also backport it as well? We did discuss whether or not the method is even useful [here](https://github.com/apache/kafka/pull/14024#discussion_r1265694650), but given that we technically do support such use cases (and can't prove that it isn't used), it'd probably be good to backport a fix for such a bad bug if possible. WDYT?
> And IMO this is a decent reason to abandon using methods to differentiate between fields and parameters
Yeah, I'm definitely convinced since I missed it as well 😆
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #14024: KAFKA-13431: Expose the original pre-transform topic partition and offset in sink records
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14024:
URL: https://github.com/apache/kafka/pull/14024#discussion_r1266406455
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
Review Comment:
> But it also seems a little ham-fisted to recommend that developers throw an exception with a message like "This connector is incompatible with this version of Kafka Connect".
Yeah, I really wanted to avoid this because connectors don't have a clean way to find out if they're configured with T/P/O mutating transformations and it doesn't make sense to be completely incompatible with older runtimes. That's why I'd included the following note after the code block:
```
Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate topic
names when deployed to older Connect runtimes that do not support this method.
```
I think adding a warning log is a good additional step.
##########
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:
I figured this was intentional to differentiate between the use of instance fields and method arguments and chose to stick with the convention used here.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/InternalSinkRecord.java:
##########
@@ -34,27 +31,12 @@ public class InternalSinkRecord extends SinkRecord {
public InternalSinkRecord(ConsumerRecord<byte[], byte[]> originalRecord, SinkRecord record) {
super(record.topic(), record.kafkaPartition(), record.keySchema(), record.key(),
- record.valueSchema(), record.value(), record.kafkaOffset(), record.timestamp(),
- record.timestampType(), record.headers());
+ record.valueSchema(), record.value(), record.kafkaOffset(), record.timestamp(),
+ record.timestampType(), record.headers(), originalRecord.topic(), originalRecord.partition(),
+ originalRecord.offset());
this.originalRecord = originalRecord;
}
- protected InternalSinkRecord(ConsumerRecord<byte[], byte[]> originalRecord, String topic,
- int partition, Schema keySchema, Object key, Schema valueSchema,
- Object value, long kafkaOffset, Long timestamp,
- TimestampType timestampType, Iterable<Header> headers) {
- super(topic, partition, keySchema, key, valueSchema, value, kafkaOffset, timestamp, timestampType, headers);
- this.originalRecord = originalRecord;
- }
-
- @Override
- public SinkRecord newRecord(String topic, Integer kafkaPartition, Schema keySchema, Object key,
Review Comment:
> Is there a reason you needed to remove this override
Hm no, I removed it because I came to the same conclusion you did about its usefulness.
> Without understanding the need to include or exclude it, i'd err on keeping what was already there.
Fair point, I've added it back with the necessary modifications (and a small unit test similar to the one for `SinkRecord`).
This bit is also relevant and while it still doesn't seem like a realistic potential case, I agree that it's probably better to not change this for now
https://github.com/apache/kafka/blob/0c6b1a4e9a94cc8d7461b78e5bb0b583dc0307d7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/WorkerErrantRecordReporter.java#L77-L80
##########
connect/api/src/test/java/org/apache/kafka/connect/sink/SinkRecordTest.java:
##########
@@ -125,4 +125,19 @@ public void shouldModifyRecordHeader() {
Header header = record.headers().lastWithName("intHeader");
assertEquals(100, (int) Values.convertToInteger(header.schema(), header.value()));
}
-}
\ No newline at end of file
+
+ @Test
+ public void shouldRetainOriginalTopicPartition() {
Review Comment:
Sounds good, no harm in having the additional test as well 👍
##########
connect/api/src/main/java/org/apache/kafka/connect/sink/SinkRecord.java:
##########
@@ -57,6 +73,105 @@ public TimestampType timestampType() {
return timestampType;
}
+ /**
+ * Get the original topic for this sink record, corresponding to the topic of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #topic()}, in order to be compatible with
+ * transformations that mutate the topic name.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalTopic;
+ * try {
+ * originalTopic = record.originalTopic();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalTopic = record.topic();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * name when deployed to older Connect runtimes.
+ *
+ * @return the topic corresponding to the Kafka record before any transformations were applied
+ *
+ * @since 3.6
+ */
+ public String originalTopic() {
+ return originalTopic;
+ }
+
+ /**
+ * Get the original topic partition for this sink record, corresponding to the topic partition of the Kafka record
+ * before any {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should
+ * be used by sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #kafkaPartition()}, in order to be compatible
+ * with transformations that mutate the topic partition value.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalKafkaPartition;
+ * try {
+ * originalKafkaPartition = record.originalKafkaPartition();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * originalKafkaPartition = record.kafkaPartition();
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Note that sink connectors that do their own offset tracking will be incompatible with SMTs that mutate the topic
+ * partition when deployed to older Connect runtimes.
+ *
+ * @return the topic partition corresponding to the Kafka record before any transformations were applied
+ *
+ * @since 3.6
+ */
+ public Integer originalKafkaPartition() {
+ return originalKafkaPartition;
+ }
+
+ /**
+ * Get the original offset for this sink record, corresponding to the offset of the Kafka record before any
+ * {@link org.apache.kafka.connect.transforms.Transformation Transformation}s were applied. This should be used by
+ * sink tasks for any internal offset tracking purposes (that are reported to the framework via
+ * {@link SinkTask#preCommit(Map)} for instance) rather than {@link #kafkaOffset()}, in order to be
+ * compatible with transformations that mutate the offset value.
+ * <p>
+ * This method was added in Apache Kafka 3.6. Sink connectors that use this method but want to maintain backward
+ * compatibility in order to be able to be deployed on older Connect runtimes should guard the call to this method
+ * with a try-catch block, since calling this method will result in a {@link NoSuchMethodException} or
+ * {@link NoClassDefFoundError} when the sink connector is deployed to Connect runtimes older than Kafka 3.6.
+ * For example:
+ * <pre>{@code
+ * String originalKafkaOffset;
+ * try {
+ * originalKafkaOffset = record.originalKafkaOffset();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
Review Comment:
Ah good point, I actually used `SourceTaskContext::transactionContext` as the reference here but that has the same issue with type existence.
> Here, where the return types are native Java types, I think the NoSuchMethodError is the only error possible here. Can you verify this?
Yup, this is correct and I've updated the Javadocs for these methods, thanks!
##########
connect/api/src/test/java/org/apache/kafka/connect/sink/SinkRecordTest.java:
##########
@@ -125,4 +125,19 @@ public void shouldModifyRecordHeader() {
Header header = record.headers().lastWithName("intHeader");
assertEquals(100, (int) Values.convertToInteger(header.schema(), header.value()));
}
-}
\ No newline at end of file
+
+ @Test
+ public void shouldRetainOriginalTopicPartition() {
Review Comment:
Sounds good, no harm in having the additional test as well 👍
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org