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/01 07:04:07 UTC
[GitHub] [kafka] yashmayya opened a new pull request, #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
yashmayya opened a new pull request, #13945:
URL: https://github.com/apache/kafka/pull/13945
- https://issues.apache.org/jira/browse/KAFKA-15121
- https://cwiki.apache.org/confluence/display/KAFKA/KIP-875%3A+First-class+offsets+support+in+Kafka+Connect#KIP875:FirstclassoffsetssupportinKafkaConnect-ConnectorAPIchanges
### 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] C0urante commented on a diff in pull request #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1261646923
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
Hmmm... wouldn't that be a pretty serious breaking change if we accidentally switched up how the JSON converter deserializes integer types? Not just for the file source connector, but for plenty of others.
It feels like it might be a better use of our time to make note of this possibility and ensure that we have sufficient unit testing in place to prevent that kind of regression (I suspect we already do but haven't verified this yet).
Of course, because things aren't interesting enough already--it turns out that there's actually two different scenarios in which tasks observe offsets for their connector. The first, which we're all familiar with, is when they query them using an [OffsetStorageReader](https://github.com/apache/kafka/blob/trunk/connect/api/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReader.java), which in distributed mode reflects the contents of the offsets topic. The second is when [SourceTask::commitRecord](https://kafka.apache.org/35/javadoc/org/apache/kafka/connect/source/SourceTask.html#commitRecord(org.apache.kafka.connect.source.SourceRecord,org.apache.kafka.clients.producer.RecordMetadata)) is invoked, which carries with it the just-ack'd `SourceRecord` instance originally provided by the task, including the original in-memory source partition and source offset, which may use types that get lost when written to and read back from the offsets topic.
I don't know if this significantly changes the conversation but it seems subtle and counterintuitive enough to bring up so that we can avoid accidentally breaking connector code that relies on this behavior.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1258640264
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,35 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified");
Review Comment:
Should we tell users that this is because stdin will be used and we don't do any offset tracking in that case?
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,35 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset != null && !offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+ }
+
+ // Let the task validate the actual value for the offset position on startup
Review Comment:
Any reason not to do this preemptively? We could at least validate that the value for the "position" key is non-null, is a numeric type, and is non-negative.
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java:
##########
@@ -147,4 +151,59 @@ public void testInvalidBatchSize() {
sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd");
assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties));
}
+
+ @Test
+ public void testAlterOffsetsStdin() {
+ sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectPartitionKey() {
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ Collections.singletonMap("invalid_partition_key", FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+
+ // null partitions are invalid
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ null,
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+ }
+
+ @Test
+ public void testAlterOffsetsMultiplePartitions() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = new HashMap<>();
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0));
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null);
+ connector.alterOffsets(sourceProperties, offsets);
Review Comment:
Can we wrap this and other successful calls to `connector::alterOffsets` in a call to `assertTrue`?
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java:
##########
@@ -147,4 +151,59 @@ public void testInvalidBatchSize() {
sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd");
assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties));
}
+
+ @Test
+ public void testAlterOffsetsStdin() {
+ sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectPartitionKey() {
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ Collections.singletonMap("invalid_partition_key", FILENAME),
Review Comment:
Nit: IMO this implies that the presence of, e.g., an `invalid_partition_key` field in a source partition is not permitted, but in reality, it looks like we do permit keys like that, as long as the `filename` field is also present in the source partition.
Perhaps we could rename this to "other_partition_key" so that it's clearer that it's not really this key's fault that the request gets rejected?
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java:
##########
@@ -147,4 +151,59 @@ public void testInvalidBatchSize() {
sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd");
assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties));
}
+
+ @Test
+ public void testAlterOffsetsStdin() {
+ sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectPartitionKey() {
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ Collections.singletonMap("invalid_partition_key", FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+
+ // null partitions are invalid
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ null,
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+ }
+
+ @Test
+ public void testAlterOffsetsMultiplePartitions() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = new HashMap<>();
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0));
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null);
+ connector.alterOffsets(sourceProperties, offsets);
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectOffsetKey() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap("invalid_offset_key", 0)
Review Comment:
Same nit as with the `testAlterOffsetsIncorrectPartitionKey` case: maybe `other_offset_key` instead?
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1259462328
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java:
##########
@@ -147,4 +151,59 @@ public void testInvalidBatchSize() {
sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd");
assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties));
}
+
+ @Test
+ public void testAlterOffsetsStdin() {
+ sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectPartitionKey() {
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ Collections.singletonMap("invalid_partition_key", FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+
+ // null partitions are invalid
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ null,
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+ }
+
+ @Test
+ public void testAlterOffsetsMultiplePartitions() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = new HashMap<>();
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0));
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null);
+ connector.alterOffsets(sourceProperties, offsets);
Review Comment:
Ah, good point. Done.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,35 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified");
Review Comment:
Makes sense, done.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,35 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset != null && !offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+ }
+
+ // Let the task validate the actual value for the offset position on startup
Review Comment:
My initial reasoning was that we probably wouldn't want to open an input stream here to read from the file and verify whether the offset position is actually valid. But yeah, no reason we can't do basic validations like checking whether it's a non-negative numeric value. I've made this change.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1259821609
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
Ah, good catch, thanks. I think it might probably be a bit friendlier if we update the task class instead to do similar parsing, WDYT? I'm okay either way, since the most common use case would be copy pasting the output from `GET /offsets` and modifying it in which case users would end up using a number rather than a string anyway.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
Review Comment:
Whoops, thanks. I forgot that we're no longer doing the single partition offset pair validation 🤦
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1259821859
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
Review Comment:
Whoops, thanks. I forgot that we're no longer doing the single partition offset pair validation 🤦
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1266275006
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,49 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
Review Comment:
Hah this seems very similar to what we've been discussing on the MM2 PR and has somewhat comparable flexibility - user friendliness tradeoffs 😅
Allowing users to modify offsets when the connector is currently configured to read from stdin could be useful in order to clean up stale offsets. However, it could be slightly misleading to accept offsets modification requests and return a successful response and message when in actuality the operation was a no-op as far as the connector's task is concerned. I'd also like to note for posterity that it isn't possible to detect past connector configurations or previous offsets in the `SourceConnector::alterOffsets` method. Both the following cases seem uncommon enough that either approach should be acceptable -
- An unfamiliar user attempts to modify offsets for a file stream source connector configured to read from stdin and expects the connector to re-produce past data.
- A user configures a file stream source connector to read from a file, then reconfigures it to read from stdin, and then wants to clean up previous offsets.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,49 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
Review Comment:
Hah this seems very similar to what we've been discussing on the MM2 PR and has somewhat comparable flexibility - user friendliness tradeoffs 😅
Allowing users to modify offsets when the connector is currently configured to read from stdin could be useful in order to clean up stale offsets. However, it could be slightly misleading to accept offsets modification requests and return a successful response and message when in actuality the operation was a no-op as far as the connector's task is concerned. I'd also like to note for posterity that it isn't possible to detect past connector configurations or previous offsets in the `SourceConnector::alterOffsets` method. Both the following cases seem uncommon enough that either approach should be acceptable -
- An unfamiliar user attempts to modify offsets for a file stream source connector configured to read from stdin and expects the connector to re-produce past data.
- A user configures a file stream source connector to read from a file, then reconfigures it to read from stdin, and then wants to clean up previous offsets.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1259882152
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
I'd prefer to leave the task parsing the same; less work on our part, and less risk of a regression in existing parts of the code base.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1260946148
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
I found something interesting when experimenting with this. When passing a request body to the `PATCH /offsets` endpoint like:
```
{
"offsets": [
{
"partition": {
"filename": "/path/to/filename"
},
"offset": {
"position": 20
}
}
]
}
```
the position `20` is deserialized to an `Integer` by Jackson (the JSON library we use in the REST layer for Connect) which seems fine because JSON doesn't have separate types for 32 bit and 64 bit numbers. So, the offsets map that is passed to `FileStreamSourceConnector::alterOffsets` by the runtime also contains `20` as an `Integer` value. I initially thought that this would cause the `FileStreamSourceTask` to fail at startup because it uses an `instanceof Long` check [here](https://github.com/apache/kafka/blob/aafbe3444354cfb0e4a8fdbd3443a63ba867c732/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceTask.java#L95C58-L95C58) (and an `Integer` value is obviously not an instance of `Long`). However, interestingly, the task did not fail and doing some debugging revealed that after the offsets are serialized and deserialized by the `JsonConverter` in `OffsetsStorageWriter` (in `Worker::modifySourceConnectorOffsets`) and `OffsetsStorageReader` respectively, the
offsets map that is retrieved by the task on startup through its context contains the position `20` as a `Long` value.
While this particular case is easily handled by simply accepting `Integer` values as valid in the `FileStreamSourceConnector::alterOffsets` method, I'm thinking we probably need to make some changes so that the offsets map passed to source connectors in their `alterOffsets` method is the same as the offsets map that connectors / tasks will retrieve via the `OffsetsStorageReader` from their context (otherwise, this could lead to some hard to debug issues in other connectors implementing the `SourceConnector::alterOffsets` method). The easiest way off the top of my head would probably would be to serialize and deserialize the offsets map using the `JsonConverter` before invoking `SourceConnector::alterOffsets`. WDYT?
Furthermore, just checking whether the offset position is an instance of `Long` (Jackson uses a `Long` if the number doesn't fit in an `Integer`) or `Integer` in the `FileStreamSourceConnector::alterOffsets` method seems sub-optimal because:
- To someone just reading through the `FileStreamSourceConnector` and `FileStreamSourceTask` classes, accepting `Integer` instances during validation but requiring `Long` instances in the actual task would look like a bug since the serialization + deserialization aspect isn't transparent.
- It's an extremely unlikely scenario, but any changes in Jackson's deser logic could break things here - for instance, if smaller numbers are deserialized into `Short`s instead of `Integer`s. Parsing using a combination of `String::valueOf` and `Long::parseLong` seems a lot more robust.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1257110479
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,40 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified");
+ }
+
+ // An empty offsets map could indicate that the offsets were reset previously or that no offsets have been committed yet (for a reset operation)
+ // - we don't need any additional validation for this case.
+ if (offsets.size() == 0) {
+ return true;
+ }
+
+ // This connector makes use of a single source partition which represents the file that it is configured to read from
+ if (offsets.size() > 1) {
+ throw new ConnectException("The " + FileStreamSourceConnector.class.getSimpleName() + " supports only a single source partition / file");
Review Comment:
Oh wow, that's a really good catch, thanks! One point to note is that in a `DELETE /offsets` request for a source connector, users don't specify the partitions / offsets (we write `null` offsets for every known partition).
Thinking about this a bit more, we shouldn't fail the request even if there are multiple partitions with non-null values. Taking the same example you used, if a user calls `GET /offsets` and copies the response body as the request body for a `PATCH /offsets` request while only modifying the offset value for file `b`, I don't think it makes sense to fail that request. Also, requests that don't even include the current source partition are also fairly harmless. I've relaxed the validation criteria in the latest patch.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #13945:
URL: https://github.com/apache/kafka/pull/13945#issuecomment-1630945143
I was toying with the idea of adding ITs for these file connectors in this very PR earlier; but after a bit of tinkering, left it for later 😄
I'd be happy to file a separate ticket for that and assign it to myself.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1256448988
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,40 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified");
+ }
+
+ // An empty offsets map could indicate that the offsets were reset previously or that no offsets have been committed yet (for a reset operation)
+ // - we don't need any additional validation for this case.
+ if (offsets.size() == 0) {
+ return true;
+ }
+
+ // This connector makes use of a single source partition which represents the file that it is configured to read from
+ if (offsets.size() > 1) {
+ throw new ConnectException("The " + FileStreamSourceConnector.class.getSimpleName() + " supports only a single source partition / file");
Review Comment:
Suppose that:
1. The connector is configured with one file `a`
2. The task commits offsets for `a`
3. The connector is reconfigured for a different file `b`
4. The task commits offsets for `b`
5. A user calls `GET /offsets` and sees both `a` and `b`
6. A user calls `DELETE /offsets` with `a`, or `a` and `b` in the same request.
I think in this situation, it is reasonable for the delete to succeed; The user clearly intends to delete the old offset, and we shouldn't force them to revert the configuration to contain `a` to delete that offset. Also, a `PATCH /offsets` which clears `a` and updates `b` in the same request should succeed.
I think it is reasonable to fail a request which updates `a`, since that is not an offset which the current configuration could produce.
So in other words: If the value is non-null, then we should validate the key. If the value is null, then any filename should be allowed, since it might have been from a previous configuration.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1261646923
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
Hmmm... wouldn't that be a pretty serious breaking change if we accidentally switched up how the JSON converter deserializes integer types? Not just for the file source connector, but for plenty of others.
It feels like it might be a better use of our time to make note of this possibility and ensure that we have sufficient unit testing in place to prevent that kind of regression (I suspect we already do but haven't verified this yet).
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1261948519
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
> Hmmm... wouldn't that be a pretty serious breaking change if we accidentally switched up how the JSON converter deserializes integer types? Not just for the file source connector, but for plenty of others.
Okay, that's fair enough, I've changed the check in `FileStreamSourceConnector::alterOffsets` to mirror the one made in the task at startup for consistency (and avoided making changes in the existing task logic). This does mean that this PR should be merged after https://github.com/apache/kafka/pull/14003 has been merged (assuming that that approach is acceptable).
> I don't know if this significantly changes the conversation but it seems subtle and counterintuitive enough to bring up so that we can avoid accidentally breaking connector code that relies on this behavior.
Hm yeah, that's definitely another interesting one to bring up - however, I'd contend that that one kinda makes sense since we're passing the `SourceRecord` itself - tasks already deal with `SourceRecord` and their offsets (and associated types) in their regular lifecycle. It would be highly confusing if the `SourceRecord` that they get in `commitRecord` doesn't match the one they dispatched to the framework via `poll`. Of course, ideally, the offsets that they read via `OffsetStorageReader` should also not have any type mismatches compared to the `SourceRecord` ones, but I don't think we'd want to (or safely could) change that at this point.
Since the offsets being altered externally would correspond to the ones that the connector / tasks read at startup, I think it makes sense to align the types across invocations to `SourceConnector::alterOffsets` and offsets queried from an `OffsetStorageReader` (and an implicit separate alignment between the `SourceRecord`'s offsets types).
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1259782359
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
Review Comment:
Shouldn't this be `continue`?
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
Review Comment:
We can remove this comment now, right?
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java:
##########
@@ -147,4 +152,102 @@ public void testInvalidBatchSize() {
sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd");
assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties));
}
+
+ @Test
+ public void testAlterOffsetsStdin() {
+ sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectPartitionKey() {
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ Collections.singletonMap("other_partition_key", FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+
+ // null partitions are invalid
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ null,
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+ }
+
+ @Test
+ public void testAlterOffsetsMultiplePartitions() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = new HashMap<>();
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0));
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null);
+ assertTrue(connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectOffsetKey() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap("other_offset_key", 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsOffsetPositionValues() {
Review Comment:
Maybe getting a little too fancy here, but it looks like we repeat the same four-line stanza several times in this test case and could reduce duplication.
I toyed with the idea of creating a function-local function since a separate `private boolean alterOffsets(Object offset)` method seemed a bit too cluttered:
```java
@Test
public void testAlterOffsetsOffsetPositionValues() {
Function<Object, Boolean> alterOffsets = offset -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
Collections.singletonMap(FILENAME_FIELD, FILENAME),
Collections.singletonMap(POSITION_FIELD, offset)
));
assertThrows(ConnectException.class, () -> alterOffsets.apply("nan"));
assertThrows(ConnectException.class, () -> alterOffsets.apply(null));
// ...
assertTrue(alterOffsets.apply(10));
}
```
Thoughts?
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
Isn't this too permissive? Based on the `testAlterOffsetsOffsetPositionValues` test case, this would allow values of both `"10"` (string) and `10` (number) for the offset. But it looks like the source task class would [only work with numbers](https://github.com/apache/kafka/blob/b8f3776f2458d0270276bd3d59db66f4c7f3748c/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceTask.java#L95-L96).
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1259827626
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java:
##########
@@ -147,4 +152,102 @@ public void testInvalidBatchSize() {
sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd");
assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties));
}
+
+ @Test
+ public void testAlterOffsetsStdin() {
+ sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectPartitionKey() {
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ Collections.singletonMap("other_partition_key", FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+
+ // null partitions are invalid
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ null,
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+ }
+
+ @Test
+ public void testAlterOffsetsMultiplePartitions() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = new HashMap<>();
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0));
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null);
+ assertTrue(connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectOffsetKey() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap("other_offset_key", 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsOffsetPositionValues() {
Review Comment:
Nice! 🚀
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java:
##########
@@ -147,4 +152,102 @@ public void testInvalidBatchSize() {
sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd");
assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties));
}
+
+ @Test
+ public void testAlterOffsetsStdin() {
+ sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectPartitionKey() {
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ Collections.singletonMap("other_partition_key", FILENAME),
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+
+ // null partitions are invalid
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
+ null,
+ Collections.singletonMap(POSITION_FIELD, 0)
+ )));
+ }
+
+ @Test
+ public void testAlterOffsetsMultiplePartitions() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = new HashMap<>();
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0));
+ offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null);
+ assertTrue(connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsIncorrectOffsetKey() {
+ Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
+ Collections.singletonMap(FILENAME_FIELD, FILENAME),
+ Collections.singletonMap("other_offset_key", 0)
+ );
+ assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
+ }
+
+ @Test
+ public void testAlterOffsetsOffsetPositionValues() {
Review Comment:
Nice! 🚀
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1261515978
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
Since the type alignment issue seemed like a broader one (i.e. not scoped to the file connectors being touched here), I've created a separate [ticket](https://issues.apache.org/jira/browse/KAFKA-15182) and [PR](https://github.com/apache/kafka/pull/14003) for it.
> it doesn't seem like a great endorsement of our API if we have to implement workarounds in the file connectors, which are the first example of the connector API that many developers see.
I'd argue that it isn't really a workaround and that the current check itself is bad. If the (de)serialization happened to use `Integer` for values that fit in a 32 bit signed type (which would be perfectly valid and is exactly what we see currently before the values are passed through the converter), the current check in the `FileStreamSourceTask` would cause it to bail.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1266275006
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,49 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
Review Comment:
Hah this seems very similar to what we've been discussing on the MM2 PR and has somewhat comparable flexibility - user friendliness tradeoffs 😅
Allowing users to modify offsets when the connector is currently configured to read from stdin could be useful in order to clean up stale offsets. However, it could be slightly misleading to accept offsets modification requests and return a successful response and message when in actuality the operation was a no-op as far as the connector's task is concerned. I'd also like to note for posterity that it isn't possible to detect past connector configurations or previous offsets in the `SourceConnector::alterOffsets` method. Both the following cases should be uncommon enough that either approach seems acceptable to me -
- An unfamiliar user attempts to modify offsets for a file stream source connector configured to read from stdin and expects the connector to re-produce past data.
- A user configures a file stream source connector to read from a file, then reconfigures it to read from stdin, and then wants to clean up previous offsets.
I think that the first case seems a bit more likely which is why I think the existing approach seems preferable, WDYT? Also, the second case can still be satisfied by temporarily reverting the configuration if the user **_really_** wants to clean up the offsets.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1266867639
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,49 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
Review Comment:
Fair enough--I think it's fine to leave as-is for now 👍
I'll merge, but depending on the outcome of the discussion around loosening partition validation for tombstones in https://github.com/apache/kafka/pull/14005, we may want to tweak this method in a follow-up to do the same.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,49 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
Review Comment:
Fair enough--I think it's fine to leave as-is for now 👍
I'll merge, but depending on the outcome of the discussion around loosening partition validation for tombstones in https://github.com/apache/kafka/pull/14005, we may want to tweak this method in a follow-up to do the same.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1261948519
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
> Hmmm... wouldn't that be a pretty serious breaking change if we accidentally switched up how the JSON converter deserializes integer types? Not just for the file source connector, but for plenty of others.
Okay, that's fair enough, I've changed the check in `FileStreamSourceConnector::alterOffsets` to mirror the one made in the task at startup for consistency (and avoided making changes in the existing task logic).
> I don't know if this significantly changes the conversation but it seems subtle and counterintuitive enough to bring up so that we can avoid accidentally breaking connector code that relies on this behavior.
Hm yeah, that's definitely another interesting one to bring up - however, I'd contend that that one kinda makes sense since we're passing the `SourceRecord` itself - tasks already deal with `SourceRecord` and their offsets (and associated types) in their regular lifecycle. It would be highly confusing if the `SourceRecord` that they get in `commitRecord` doesn't match the one they dispatched to the framework via `poll`. Of course, ideally, the offsets that they read via `OffsetStorageReader` should also not have any type mismatches compared to the `SourceRecord` ones, but I don't think we'd want to (or safely could) change that at this point.
Since the offsets being altered externally would correspond to the ones that the connector / tasks read at startup, I think it makes sense to align the types across invocations to `SourceConnector::alterOffsets` and offsets queried from an `OffsetStorageReader` (and an implicit separate alignment between the `SourceRecord`'s offsets types).
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante merged PR #13945:
URL: https://github.com/apache/kafka/pull/13945
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1262708616
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
💯 Sounds good, thanks for thinking this through.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
💯 Sounds good, thanks for thinking this through.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1265653816
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,49 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
Review Comment:
Sorry! Just had one more thought about this case.
If the connector has been reconfigured from using a file to using stdin, doesn't it still make sense to allow users to interact with offsets for the file that it was using in the past? This is especially relevant for the [total reset case](https://github.com/apache/kafka/pull/13945#discussion_r1257110479) that you noted in the earlier discussion with Greg.
--
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 #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13945:
URL: https://github.com/apache/kafka/pull/13945#discussion_r1261413255
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}
+ @Override
+ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> offsets) {
+ AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig);
+ String filename = config.getString(FILE_CONFIG);
+ if (filename == null || filename.isEmpty()) {
+ // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked
+ throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
+ "This is because stdin is used for input and offsets are not tracked.");
+ }
+
+ // This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
+ // However, there could also be source partitions from previous configurations of the connector.
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
+ Map<String, ?> partition = partitionOffset.getKey();
+ if (partition == null) {
+ throw new ConnectException("Partition objects cannot be null");
+ }
+
+ if (!partition.containsKey(FILENAME_FIELD)) {
+ throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
+ }
+
+ Map<String, ?> offset = partitionOffset.getValue();
+ // null offsets are allowed and represent a deletion of offsets for a partition
+ if (offset == null) {
+ return true;
+ }
+
+ if (!offset.containsKey(POSITION_FIELD)) {
+ throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
+ }
+
+ // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value
+ try {
+ long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD)));
Review Comment:
Ah, nice catch! I noticed the discrepancy in numeric types while working on [KAFKA-15177](https://issues.apache.org/jira/browse/KAFKA-15177) but hadn't even considered the possibility of aligning the types across invocations of `alterOffsets` and `OffsetStorageReader::offset`/`OffsetStorageReader::offsets`.
I think re-deserializing the offsets before passing them to `alterOffsets` is a great idea. Unless the request body is gigantic there shouldn't be serious performance concerns, and it also acts as a nice preflight check to ensure that the offsets can be successfully propagated to the connector's tasks through the offsets topic.
I still don't love permitting string types for the connector's `position` offset values--it doesn't seem like a great endorsement of our API if we have to implement workarounds in the file connectors, which are the first example of the connector API that many developers see.
--
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