Improve WatchService error handling & logging (#153)

* Improve WatchService error handling & logging

* Add license header

* Revert cleanup log level to debug

Author: aymkhalil

agent/build.gradle

@@ -12,6 +12,7 @@ dependencies {
     implementation("org.apache.avro:avro:${avroVersion}")
     compileOnly("org.slf4j:slf4j-api:${slf4jVersion}")
     testImplementation("org.junit-pioneer:junit-pioneer:1.4.2")
+    testImplementation "com.datastax.oss:dsbulk-tests:${dsbulkVersion}"
 
     implementation("${pulsarGroup}:pulsar-client:${pulsarVersion}")
 }

agent/src/main/java/com/datastax/oss/cdc/agent/AbstractDirectoryWatcher.java

@@ -36,13 +36,13 @@ public abstract class AbstractDirectoryWatcher {
     private final WatchService watchService;
     private final Duration pollInterval;
     private final Path directory;
-    private final Set<WatchEvent.Kind<Path>> kinds;
+    private final Set<WatchEvent.Kind<?>> kinds;
 
-    public AbstractDirectoryWatcher(Path directory, Duration pollInterval, Set<WatchEvent.Kind<Path>> kinds) throws IOException {
+    public AbstractDirectoryWatcher(Path directory, Duration pollInterval, Set<WatchEvent.Kind<?>> kinds) throws IOException {
         this(FileSystems.getDefault().newWatchService(), directory, pollInterval, kinds);
     }
 
-    AbstractDirectoryWatcher(WatchService watchService, Path directory, Duration pollInterval, Set<WatchEvent.Kind<Path>> kinds) throws IOException {
+    AbstractDirectoryWatcher(WatchService watchService, Path directory, Duration pollInterval, Set<WatchEvent.Kind<?>> kinds) throws IOException {
         this.watchService = watchService;
         this.pollInterval = pollInterval;
         this.directory = directory;
@@ -55,16 +55,28 @@ public void poll() throws InterruptedException, IOException {
         WatchKey key = watchService.poll(pollInterval.toMillis(), TimeUnit.MILLISECONDS);
 
         if (key != null) {
-            for (WatchEvent<?> event : key.pollEvents()) {
-                Path relativePath = (Path) event.context();
-                Path absolutePath = directory.resolve(relativePath);
+            try {
+                for (WatchEvent<?> event : key.pollEvents()) {
+                    if (event.kind() == java.nio.file.StandardWatchEventKinds.OVERFLOW) {
+                        log.warn("Overflow event detected: kind={}, context={}, count={}",
+                                event.kind(), event.context(), event.count());
+                        continue;
+                    }
+                    Path relativePath = (Path) event.context();
+                    Path absolutePath = directory.resolve(relativePath);
 
-                if (kinds.contains(event.kind())) {
-                    log.debug("Detected new commitlog file={}", absolutePath);
-                    handleEvent(event, absolutePath);
+                    if (kinds.contains(event.kind())) {
+                        log.debug("Detected new commitlog file={}", absolutePath);
+                        handleEvent(event, absolutePath);
+                    }
+                }
+            } catch (Exception ex) {
+                log.error("Error while handling WatchKey", ex);
+            } finally {
+                if (!key.reset()) {
+                    log.warn("WatchKey is no longer valid");
                 }
             }
-            key.reset();
         }
     }
 

agent/src/main/java/com/datastax/oss/cdc/agent/CommitLogProcessor.java

@@ -28,6 +28,7 @@
 
 import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
 import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
+import static java.nio.file.StandardWatchEventKinds.OVERFLOW;
 
 /**
  * Detect and read commitlogs files in the cdc_raw directory.
@@ -36,7 +37,7 @@
 public class CommitLogProcessor extends AbstractProcessor implements AutoCloseable {
     private static final String NAME = "Commit Log Processor";
 
-    private static final Set<WatchEvent.Kind<Path>> watchedEvents = Stream.of(ENTRY_CREATE, ENTRY_MODIFY).collect(Collectors.toSet());
+    private static final Set<WatchEvent.Kind<?>> watchedEvents = Stream.of(ENTRY_CREATE, ENTRY_MODIFY, OVERFLOW).collect(Collectors.toSet());
 
     private final AbstractDirectoryWatcher newCommitLogWatcher;
     private final CommitLogTransfer commitLogTransfer;

agent/src/main/java/com/datastax/oss/cdc/agent/CommitLogReaderService.java

@@ -300,7 +300,7 @@ public void finish(TaskStatus taskStatus, int lastSentPosition) {
         }
 
         public void cleanup(TaskStatus status) {
-            log.debug("Cleanup task={}", this, status);
+            log.debug("Cleanup task={}, status={}", this, status);
             File file = getFile();
             switch (status) {
                 case SUCCESS:

agent/src/test/java/com/datastax/oss/cdc/agent/AbstractDirectoryWatcherTest.java

@@ -0,0 +1,97 @@
+/**
+ * Copyright DataStax, Inc 2021.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.cdc.agent;
+
+import com.datastax.oss.dsbulk.tests.utils.FileUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.WatchEvent;
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
+import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
+import static java.nio.file.StandardWatchEventKinds.OVERFLOW;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class AbstractDirectoryWatcherTest {
+
+    private Path cdcDir;
+
+    @BeforeEach
+    public void init() throws IOException {
+        cdcDir = Files.createTempDirectory("cdc");
+    }
+
+    @AfterEach
+    void deleteTempDirs() {
+        if (cdcDir != null && Files.exists(cdcDir)) {
+            FileUtils.deleteDirectory(cdcDir);
+        }
+    }
+
+    @Test
+    public void testPoll() throws InterruptedException, IOException {
+        // given
+        Set<WatchEvent.Kind<?>> watchedEvents = Stream.of(ENTRY_CREATE, ENTRY_MODIFY, OVERFLOW).collect(Collectors.toSet());
+        final BlockingQueue<File> commitLogQueue = new LinkedBlockingQueue<>();
+
+        AbstractDirectoryWatcher watcher = new AbstractDirectoryWatcher(cdcDir,
+                Duration.ofSeconds(20), // avoid flakiness, usually completes in ~10 seconds
+                watchedEvents) {
+            @Override
+            void handleEvent(WatchEvent<?> event, Path path) {
+                commitLogQueue.add(path.toFile());
+            }
+        };
+        File commitLog = new File(cdcDir.toFile(), "CommitLog-6-1.log");
+        commitLog.createNewFile();
+
+        // when
+        watcher.poll(); // can block up to 20 seconds
+
+        // then
+        assertEventHandled(commitLogQueue, commitLog);
+
+        // write another file
+        File commitLog2 = new File(cdcDir.toFile(), "CommitLog-6-2.log");
+        commitLog2.createNewFile();
+
+        // when
+        watcher.poll(); // can block up to 20 seconds
+
+        // then
+        assertEventHandled(commitLogQueue, commitLog2);
+    }
+
+    private void assertEventHandled(BlockingQueue<File> commitLogQueue, File expectedFile)  {
+        File actualFile = commitLogQueue.poll();
+        assertNotNull(actualFile);
+        assertEquals(commitLogQueue.size(), 0);
+        assertEquals(expectedFile, actualFile);
+    }
+}