This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new fcedbbc1c [ISSUE #4341] WebHookConfig failed to load after being
inserted (#4344)
fcedbbc1c is described below
commit fcedbbc1caa01b5f441c24c61ddcce3d1b29d924
Author: Pil0tXia <[email protected]>
AuthorDate: Fri Aug 25 09:39:42 2023 +0800
[ISSUE #4341] WebHookConfig failed to load after being inserted (#4344)
* chore: add final to cacheWebHookConfig
* fix: write cache only when MODIFY to avoid write empty config when CREATE
* chore: correct log grammar mistake
* feat: add try-catch and comments
* fix: Wait for the notification of file write completion before
initializing the cache
* fix: use a shared latch
* doc: add comment for cacheInit to explain CountDownLatch
* feat: add object lock to ensure the atomicity of countDown in case of
concurrency
* feat: use synchronized instead of CountDownLatch
* fix: expand synchronize zone to avoid file deleted before cacheInit
* fix checkstyle
---
.../webhook/admin/FileWebHookConfigOperation.java | 46 ++++++++++++----------
.../webhook/api/common/SharedLatchHolder.java | 25 ++++++++++++
.../receive/storage/WebhookFileListener.java | 46 ++++++++++++++--------
3 files changed, 80 insertions(+), 37 deletions(-)
diff --git
a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
index fd8535a53..db222905c 100644
---
a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
+++
b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.webhook.api.WebHookConfig;
import org.apache.eventmesh.webhook.api.WebHookConfigOperation;
import org.apache.eventmesh.webhook.api.WebHookOperationConstant;
+import org.apache.eventmesh.webhook.api.common.SharedLatchHolder;
import org.apache.eventmesh.webhook.api.utils.ClassUtils;
import java.io.BufferedReader;
@@ -88,7 +89,7 @@ public class FileWebHookConfigOperation implements
WebHookConfigOperation {
final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
if (!webhookConfigFile.exists()) {
if (log.isErrorEnabled()) {
- log.error("webhookConfig {} is not existed",
webHookConfig.getCallbackPath());
+ log.error("webhookConfig {} does not exist",
webHookConfig.getCallbackPath());
}
return 0;
}
@@ -97,14 +98,16 @@ public class FileWebHookConfigOperation implements
WebHookConfigOperation {
@Override
public Integer deleteWebHookConfig(final WebHookConfig webHookConfig) {
- final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
- if (!webhookConfigFile.exists()) {
- if (log.isErrorEnabled()) {
- log.error("webhookConfig {} is not existed",
webHookConfig.getCallbackPath());
+ synchronized (SharedLatchHolder.lock) {
+ final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
+ if (!webhookConfigFile.exists()) {
+ if (log.isErrorEnabled()) {
+ log.error("webhookConfig {} does not exist",
webHookConfig.getCallbackPath());
+ }
+ return 0;
}
- return 0;
+ return webhookConfigFile.delete() ? 1 : 0;
}
- return webhookConfigFile.delete() ? 1 : 0;
}
/**
@@ -115,7 +118,7 @@ public class FileWebHookConfigOperation implements
WebHookConfigOperation {
final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
if (!webhookConfigFile.exists()) {
if (log.isErrorEnabled()) {
- log.error("webhookConfig {} is not existed",
webHookConfig.getCallbackPath());
+ log.error("webhookConfig {} does not exist",
webHookConfig.getCallbackPath());
}
return null;
}
@@ -131,7 +134,7 @@ public class FileWebHookConfigOperation implements
WebHookConfigOperation {
final File manuDir = new File(manuDirPath);
if (!manuDir.exists()) {
if (log.isWarnEnabled()) {
- log.warn("webhookConfig dir {} is not existed", manuDirPath);
+ log.warn("webhookConfig dir {} does not exist", manuDirPath);
}
return new ArrayList<>();
}
@@ -164,7 +167,7 @@ public class FileWebHookConfigOperation implements
WebHookConfigOperation {
}
} catch (IOException e) {
if (log.isErrorEnabled()) {
- log.error("get webhook from file {} error",
webhookConfigFile.getPath(), e);
+ log.error("get webHookConfig from file {} error",
webhookConfigFile.getPath(), e);
}
return null;
}
@@ -173,18 +176,21 @@ public class FileWebHookConfigOperation implements
WebHookConfigOperation {
}
public static boolean writeToFile(final File webhookConfigFile, final
WebHookConfig webHookConfig) {
- try (FileOutputStream fos = new FileOutputStream(webhookConfigFile);
- BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fos,
StandardCharsets.UTF_8))) {
- // lock this file, and will auto release after fos close
- fos.getChannel().lock();
-
bw.write(Objects.requireNonNull(JsonUtils.toJSONString(webHookConfig)));
- } catch (IOException e) {
- if (log.isErrorEnabled()) {
- log.error("write webhookConfig {} to file error",
webHookConfig.getCallbackPath());
+ // Wait for the previous cacheInit to complete in case of concurrency
+ synchronized (SharedLatchHolder.lock) {
+ try (FileOutputStream fos = new
FileOutputStream(webhookConfigFile);
+ BufferedWriter bw = new BufferedWriter(new
OutputStreamWriter(fos, StandardCharsets.UTF_8))) {
+ // Lock this file to prevent concurrent modification and it
will be automatically unlocked when fos closes
+ fos.getChannel().lock();
+
bw.write(Objects.requireNonNull(JsonUtils.toJSONString(webHookConfig)));
+ } catch (IOException e) {
+ if (log.isErrorEnabled()) {
+ log.error("write webhookConfig {} to file error",
webHookConfig.getCallbackPath());
+ }
+ return false;
}
- return false;
+ return true;
}
- return true;
}
private String getWebhookConfigManuDir(final WebHookConfig webHookConfig) {
diff --git
a/eventmesh-webhook/eventmesh-webhook-api/src/main/java/org/apache/eventmesh/webhook/api/common/SharedLatchHolder.java
b/eventmesh-webhook/eventmesh-webhook-api/src/main/java/org/apache/eventmesh/webhook/api/common/SharedLatchHolder.java
new file mode 100644
index 000000000..c59ff54b3
--- /dev/null
+++
b/eventmesh-webhook/eventmesh-webhook-api/src/main/java/org/apache/eventmesh/webhook/api/common/SharedLatchHolder.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.webhook.api.common;
+
+public class SharedLatchHolder {
+
+ // secure the execution sequence of writeToFile and cacheInit
+ public static final Object lock = new Object();
+
+}
diff --git
a/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/storage/WebhookFileListener.java
b/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/storage/WebhookFileListener.java
index 5461cdcbb..8d42c9f19 100644
---
a/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/storage/WebhookFileListener.java
+++
b/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/storage/WebhookFileListener.java
@@ -24,6 +24,7 @@ import static
java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.webhook.api.WebHookConfig;
import org.apache.eventmesh.webhook.api.WebHookOperationConstant;
+import org.apache.eventmesh.webhook.api.common.SharedLatchHolder;
import java.io.BufferedReader;
import java.io.File;
@@ -51,7 +52,7 @@ public class WebhookFileListener {
private final transient Set<String> pathSet = new LinkedHashSet<>(); //
monitored subdirectory
private final transient Map<WatchKey, String> watchKeyPathMap = new
ConcurrentHashMap<>(); // WatchKey's path
private transient String filePath;
- private transient Map<String, WebHookConfig> cacheWebHookConfig;
+ private final transient Map<String, WebHookConfig> cacheWebHookConfig;
public WebhookFileListener(final String filePath, final Map<String,
WebHookConfig> cacheWebHookConfig) {
this.filePath = WebHookOperationConstant.getFilePath(filePath);
@@ -90,20 +91,21 @@ public class WebhookFileListener {
}
/**
- * Read the file and cache it in map
+ * Read the file and cache it in local map for manufacturers webhook
payload delivery
+ * <p>
+ * A shared lock is used to ensure that this method should be invoked
after the {@code webhookConfigFile} is written completely
+ * by {@code
org.apache.eventmesh.webhook.admin.FileWebHookConfigOperation#writeToFile} when
multiple modify events are triggered.
*
* @param webhookConfigFile webhookConfigFile
*/
private void cacheInit(final File webhookConfigFile) {
final StringBuilder fileContent = new StringBuilder();
- try (BufferedReader br =
Files.newBufferedReader(Paths.get(webhookConfigFile.getAbsolutePath()),
- StandardCharsets.UTF_8)) {
+ try (BufferedReader br =
Files.newBufferedReader(Paths.get(webhookConfigFile.getAbsolutePath()),
StandardCharsets.UTF_8)) {
while (br.ready()) {
fileContent.append(br.readLine());
}
-
} catch (IOException e) {
- log.error("cacheInit failed", e);
+ log.error("cacheInit buffer read failed", e);
}
final WebHookConfig webHookConfig =
JsonUtils.parseObject(fileContent.toString(), WebHookConfig.class);
cacheWebHookConfig.put(webhookConfigFile.getName(), webHookConfig);
@@ -143,36 +145,46 @@ public class WebhookFileListener {
WatchKey key = null;
try {
assert service != null;
+ // The code will block here until a file system event
occurs
key = service.take();
} catch (InterruptedException e) {
log.error("Interrupted", e);
}
assert key != null;
+ // A newly created config file will be captured for two
events, ENTRY_CREATE and ENTRY_MODIFY
for (final WatchEvent<?> event : key.pollEvents()) {
final String flashPath = watchKeyPathMap.get(key);
// manufacturer change
- final String path =
flashPath.concat("/").concat(event.context().toString());
+ final String path =
flashPath.concat(WebHookOperationConstant.FILE_SEPARATOR).concat(event.context().toString());
final File file = new File(path);
- if (ENTRY_CREATE == event.kind() || ENTRY_MODIFY ==
event.kind()) {
- if (file.isFile()) {
- cacheInit(file);
- } else {
+ // Wait for file write/delete completion before
initializing the cache
+ synchronized (SharedLatchHolder.lock) {
+ if (file.isDirectory() && (ENTRY_CREATE ==
event.kind() || ENTRY_MODIFY == event.kind())) {
+ // If it is a folder, re-register the listener
try {
key = Paths.get(path).register(service,
ENTRY_CREATE, ENTRY_MODIFY, ENTRY_DELETE);
watchKeyPathMap.put(key, path);
} catch (IOException e) {
log.error("registerWatchKey failed", e);
}
- }
- } else if (ENTRY_DELETE == event.kind()) {
- if (file.isDirectory()) {
- watchKeyPathMap.remove(key);
- } else {
- deleteConfig(file);
+ } else if (file.isFile() && ENTRY_MODIFY ==
event.kind()) {
+ // If it is a file, cache it only when it is
modified to wait for complete file writes
+ try {
+ cacheInit(file);
+ } catch (Exception e) {
+ log.error("cacheInit failed", e);
+ }
+ } else if (ENTRY_DELETE == event.kind()) {
+ if (file.isDirectory()) {
+ watchKeyPathMap.remove(key);
+ } else {
+ deleteConfig(file);
+ }
}
}
}
+ // Reset the WatchKey to receive subsequent file system events
if (!key.reset()) {
break;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
