From ae0f37955c13dd85f13997bd25b5300e07d61fe2 Mon Sep 17 00:00:00 2001
From: Anatolii Popov <anatolii.popov@aiven.io>
Date: Tue, 2 Nov 2021 15:08:58 +0200
Subject: [PATCH] Fixed leak of inotify instances.

---
 .../kafka/auth/AivenAclAuthorizerV2.java      | 21 ++++++++++++-------
 1 file changed, 13 insertions(+), 8 deletions(-)

diff --git a/src/main/java/io/aiven/kafka/auth/AivenAclAuthorizerV2.java b/src/main/java/io/aiven/kafka/auth/AivenAclAuthorizerV2.java
index 976b3b1..1da5b76 100644
--- a/src/main/java/io/aiven/kafka/auth/AivenAclAuthorizerV2.java
+++ b/src/main/java/io/aiven/kafka/auth/AivenAclAuthorizerV2.java
@@ -68,19 +68,13 @@ public class AivenAclAuthorizerV2 implements Authorizer {
     private File configFile;
     private AuditorAPI auditor;
     private boolean logDenials;
-    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
-    private final WatchService watchService;
+    private ScheduledExecutorService scheduledExecutorService;
+    private volatile WatchService watchService;
     private final AtomicReference<VerdictCache> cacheReference = new AtomicReference<>();
 
     private AivenAclAuthorizerConfig config;
 
     public AivenAclAuthorizerV2() {
-        try {
-            watchService = FileSystems.getDefault().newWatchService();
-        } catch (final IOException e) {
-            LOGGER.error("Failed to initialize WatchService", e);
-            throw new RuntimeException(e);
-        }
     }
 
     @Override
@@ -93,6 +87,8 @@ public void configure(final java.util.Map<String, ?> configs) {
         final AuthorizerServerInfo serverInfo) {
         auditor = config.getAuditor();
         logDenials = config.logDenials();
+        watchService = initializeWatchService();
+        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
 
         configFile = config.getConfigFile();
         final AclJsonReader jsonReader = new AclJsonReader(configFile.toPath());
@@ -128,6 +124,15 @@ public void configure(final java.util.Map<String, ?> configs) {
             ));
     }
 
+    private WatchService initializeWatchService() {
+        try {
+            return FileSystems.getDefault().newWatchService();
+        } catch (final IOException e) {
+            LOGGER.error("Failed to initialize WatchService", e);
+            throw new RuntimeException(e);
+        }
+    }
+
     private WatchKey subscribeToAclChanges(final File configFile) {
         try {
             return configFile.toPath().toAbsolutePath().getParent()