Created dedicated timeout for task

Author: riferrei

src/main/java/tutorial/buildon/aws/streaming/kafka/MyFirstKafkaConnectorConfig.java

@@ -27,6 +27,10 @@ public MyFirstKafkaConnectorConfig(final Map<?, ?> originalProps) {
     private static final String SECOND_NONREQUIRED_PARAM_DOC = "This is the 2nd non-required parameter";
     private static final String SECOND_NONREQUIRED_PARAM_DEFAULT = "bar";
 
+    public static final String TASK_SLEEP_TIMEOUT_CONFIG = "task.sleep.timeout";
+    private static final String TASK_SLEEP_TIMEOUT_DOC = "Sleep timeout used by tasks during each poll";
+    private static final int TASK_SLEEP_TIMEOUT_DEFAULT = 5000;
+
     public static final String MONITOR_THREAD_TIMEOUT_CONFIG = "monitor.thread.timeout";
     private static final String MONITOR_THREAD_TIMEOUT_DOC = "Timeout used by the monitoring thread";
     private static final int MONITOR_THREAD_TIMEOUT_DEFAULT = 10000;
@@ -62,6 +66,12 @@ private static void addParams(final ConfigDef configDef) {
             SECOND_NONREQUIRED_PARAM_DEFAULT,
             Importance.HIGH,
             SECOND_NONREQUIRED_PARAM_DOC)
+        .define(
+            TASK_SLEEP_TIMEOUT_CONFIG,
+            Type.INT,
+            TASK_SLEEP_TIMEOUT_DEFAULT,
+            Importance.HIGH,
+            TASK_SLEEP_TIMEOUT_DOC)
         .define(
             MONITOR_THREAD_TIMEOUT_CONFIG,
             Type.INT,

src/main/java/tutorial/buildon/aws/streaming/kafka/MyFirstKafkaConnectorTask.java

@@ -27,7 +27,7 @@ public class MyFirstKafkaConnectorTask extends SourceTask {
     private final Logger log = LoggerFactory.getLogger(MyFirstKafkaConnectorTask.class);
 
     private MyFirstKafkaConnectorConfig config;
-    private int monitorThreadTimeout;
+    private int taskSleepTimeout;
     private List<String> sources;
     private Schema recordSchema;
 
@@ -39,7 +39,7 @@ public String version() {
     @Override
     public void start(Map<String, String> properties) {
         config = new MyFirstKafkaConnectorConfig(properties);
-        monitorThreadTimeout = config.getInt(MONITOR_THREAD_TIMEOUT_CONFIG);
+        taskSleepTimeout = config.getInt(TASK_SLEEP_TIMEOUT_CONFIG);
         String sourcesStr = properties.get("sources");
         sources = Arrays.asList(sourcesStr.split(","));
         recordSchema = SchemaBuilder.struct()
@@ -51,7 +51,7 @@ public void start(Map<String, String> properties) {
 
     @Override
     public List<SourceRecord> poll() throws InterruptedException {
-        Thread.sleep(monitorThreadTimeout / 2);
+        Thread.sleep(taskSleepTimeout);
         List<SourceRecord> records = new ArrayList<>();
         for (String source : sources) {
             log.info("Polling data from the source '" + source + "'");