[GitHub] [kafka] yashmayya commented on a diff in pull request #13945: KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector

["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