yashmayya commented on code in PR #12450:
URL: https://github.com/apache/kafka/pull/12450#discussion_r940880586
##########
connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSinkConnectorTest.java:
##########
@@ -74,16 +73,25 @@ public void testSinkTasks() {
@Test
public void testSinkTasksStdout() {
- sinkProperties.remove(FileStreamSourceConnector.FILE_CONFIG);
+ sinkProperties.remove(FileStreamSinkConnector.FILE_CONFIG);
connector.start(sinkProperties);
List<Map<String, String>> taskConfigs = connector.taskConfigs(1);
assertEquals(1, taskConfigs.size());
-
assertNull(taskConfigs.get(0).get(FileStreamSourceConnector.FILE_CONFIG));
+
assertNull(taskConfigs.get(0).get(FileStreamSinkConnector.FILE_CONFIG));
}
@Test
public void testTaskClass() {
connector.start(sinkProperties);
assertEquals(FileStreamSinkTask.class, connector.taskClass());
}
+
+ @Test
+ public void testConnectorConfigsPropagateToTaskConfigs() {
Review Comment:
Sure, makes sense, I've added comments in the source/sink tests for this. I
don't feel like it's really necessary to add it in the connectors' `start`
methods since we aren't really doing anything convoluted there as such that
needs explaining. However, I'd be okay with adding it there as well if you feel
strongly about it.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java:
##########
@@ -16,40 +16,35 @@
*/
package org.apache.kafka.connect.file;
-import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
-import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.source.SourceConnector;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
- * Very simple connector that works with the console. This connector supports
both source and
- * sink modes via its 'mode' setting.
+ * Very simple source connector that works with stdin or a file.
*/
public class FileStreamSourceConnector extends SourceConnector {
- public static final String TOPIC_CONFIG = "topic";
- public static final String FILE_CONFIG = "file";
- public static final String TASK_BATCH_SIZE_CONFIG = "batch.size";
- public static final int DEFAULT_TASK_BATCH_SIZE = 2000;
+ static final String TOPIC_CONFIG = "topic";
+ static final String FILE_CONFIG = "file";
+ static final String TASK_BATCH_SIZE_CONFIG = "batch.size";
- private static final ConfigDef CONFIG_DEF = new ConfigDef()
+ static final int DEFAULT_TASK_BATCH_SIZE = 2000;
Review Comment:
Hm alright, I guess I hadn't considered that - I've reverted these changes.
##########
connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSinkConnector.java:
##########
@@ -25,21 +24,19 @@
import org.apache.kafka.connect.sink.SinkConnector;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
- * Very simple connector that works with the console. This connector supports
both source and
- * sink modes via its 'mode' setting.
+ * Very simple sink connector that works with stdout or a file.
*/
public class FileStreamSinkConnector extends SinkConnector {
- public static final String FILE_CONFIG = "file";
- private static final ConfigDef CONFIG_DEF = new ConfigDef()
+ static final String FILE_CONFIG = "file";
+ static final ConfigDef CONFIG_DEF = new ConfigDef()
Review Comment:
Reverted the public -> package-private for `FILE_CONFIG` but we need
`CONFIG_DEF` to be at least package-private since it is now being used in the
task class as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
