You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2023/10/31 06:48:16 UTC
(eventmesh) branch master updated: [ISSUE #4502] Optimizing Log Printing Using the LogUtils
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 55865a660 [ISSUE #4502] Optimizing Log Printing Using the LogUtils
55865a660 is described below
commit 55865a66046f6461ebc4e80767d6bb47e8753bb0
Author: scwlkq <89...@users.noreply.github.com>
AuthorDate: Tue Oct 31 14:48:10 2023 +0800
[ISSUE #4502] Optimizing Log Printing Using the LogUtils
* Change the log mode to LogUtils
* check code style
* recheck
* fix some code style inappropriate places
* temp store
* fix merge conflict
* final check
* fix some field sequence problems and redundant indention problems
* Remove unnecessary parameters from the log
* fix merge conflict
* fix merge conflict
* fix merge conflict
* correct code style
* fix ci
* add missing {} in some log statements
* fix ci
---
.../eventmesh/common/file/WatchFileManager.java | 10 +-
.../eventmesh/common/file/WatchFileTask.java | 10 +-
.../eventmesh/common/protocol/tcp/codec/Codec.java | 13 +--
.../connector/jdbc/connection/JdbcConnection.java | 35 ++-----
.../source/dialect/cdc/mysql/MysqlCdcEngine.java | 4 +-
.../source/dialect/mysql/MysqlDatabaseDialect.java | 17 +---
.../WorkflowAsyncPublishInstance.java | 5 +-
.../grpc/sub/CloudEventsAsyncSubscribe.java | 5 +-
.../grpc/sub/CloudEventsSubscribeReply.java | 5 +-
.../grpc/sub/EventMeshAsyncSubscribe.java | 5 +-
.../grpc/sub/EventMeshSubscribeBroadcast.java | 5 +-
.../grpc/sub/EventMeshSubscribeReply.java | 5 +-
.../grpc/sub/WorkflowExpressAsyncSubscribe.java | 9 +-
.../grpc/sub/WorkflowOrderAsyncSubscribe.java | 9 +-
.../grpc/sub/WorkflowPaymentAsyncSubscribe.java | 9 +-
.../grpc/sub/app/controller/SubController.java | 5 +-
.../eventmesh/grpc/sub/app/service/SubService.java | 25 ++---
.../eventmeshmessage/AsyncSyncRequestInstance.java | 11 +-
.../pub/eventmeshmessage/SyncRequestInstance.java | 7 +-
.../http/demo/sub/controller/SubController.java | 12 +--
.../http/demo/sub/service/SubService.java | 21 ++--
.../tcp/demo/pub/cloudevents/AsyncPublish.java | 5 +-
.../tcp/demo/pub/cloudevents/SyncRequest.java | 9 +-
.../demo/pub/eventmeshmessage/AsyncPublish.java | 5 +-
.../eventmeshmessage/AsyncPublishBroadcast.java | 5 +-
.../tcp/demo/pub/eventmeshmessage/SyncRequest.java | 9 +-
.../tcp/demo/sub/cloudevents/AsyncSubscribe.java | 5 +-
.../tcp/demo/sub/cloudevents/SyncResponse.java | 5 +-
.../demo/sub/eventmeshmessage/AsyncSubscribe.java | 5 +-
.../eventmeshmessage/AsyncSubscribeBroadcast.java | 5 +-
.../demo/sub/eventmeshmessage/SyncResponse.java | 5 +-
.../handler/RedirectClientByIpPortHandler.java | 2 +-
.../admin/handler/RedirectClientByPathHandler.java | 2 +-
.../handler/RedirectClientBySubSystemHandler.java | 22 ++--
.../admin/handler/RejectAllClientHandler.java | 5 +-
.../admin/handler/RejectClientByIpPortHandler.java | 2 +-
.../handler/RejectClientBySubSystemHandler.java | 2 +-
.../admin/handler/ShowClientBySystemHandler.java | 5 +-
.../eventmesh/runtime/boot/AbstractHTTPServer.java | 35 ++-----
.../runtime/boot/AbstractRemotingServer.java | 38 ++++---
.../eventmesh/runtime/boot/AbstractTCPServer.java | 15 ++-
.../runtime/boot/EventMeshHTTPServer.java | 18 +---
.../eventmesh/runtime/boot/EventMeshServer.java | 21 ++--
.../eventmesh/runtime/boot/EventMeshStartup.java | 10 +-
.../eventmesh/runtime/boot/EventMeshTCPServer.java | 18 +---
.../protocol/grpc/consumer/ConsumerManager.java | 31 ++----
.../protocol/grpc/consumer/EventMeshConsumer.java | 29 ++----
.../consumer/consumergroup/StreamTopicConfig.java | 7 +-
.../grpc/processor/SubscribeProcessor.java | 17 ++--
.../http/consumer/HttpClientGroupMapping.java | 88 ++++++----------
.../http/processor/BatchSendMessageProcessor.java | 12 +--
.../processor/BatchSendMessageV2Processor.java | 5 +-
.../http/processor/CreateTopicProcessor.java | 7 +-
.../http/processor/DeleteTopicProcessor.java | 7 +-
.../protocol/http/processor/HandlerService.java | 9 +-
.../http/processor/HeartBeatProcessor.java | 23 ++---
.../processor/LocalSubscribeEventProcessor.java | 35 +++----
.../processor/LocalUnSubscribeEventProcessor.java | 30 ++----
.../http/processor/QuerySubscriptionProcessor.java | 7 +-
.../http/processor/ReplyMessageProcessor.java | 9 +-
.../http/processor/SendAsyncEventProcessor.java | 45 +++-----
.../http/processor/SendAsyncMessageProcessor.java | 9 +-
.../processor/SendAsyncRemoteEventProcessor.java | 41 +++-----
.../http/processor/SendSyncMessageProcessor.java | 57 ++++-------
.../http/processor/SubscribeProcessor.java | 32 ++----
.../http/processor/UnSubscribeProcessor.java | 32 +++---
.../protocol/http/push/AsyncHTTPPushRequest.java | 31 +++---
.../tcp/client/group/ClientGroupWrapper.java | 36 ++-----
.../client/group/ClientSessionGroupMapping.java | 5 +-
.../dispatch/FreePriorityDispatchStrategy.java | 15 +--
.../tcp/client/processor/HelloProcessor.java | 2 +-
.../tcp/client/processor/SubscribeProcessor.java | 7 +-
.../client/recommend/EventMeshRecommendImpl.java | 85 +++++-----------
.../runtime/metrics/http/HTTPMetricsServer.java | 113 +++++++++------------
.../eventmesh/runtime/util/EventMeshUtil.java | 22 ++--
.../apache/eventmesh/runtime/util/WebhookUtil.java | 11 +-
.../runtime/client/common/RequestContext.java | 5 +-
.../runtime/client/impl/PubClientImpl.java | 27 ++---
.../runtime/client/impl/SubClientImpl.java | 19 ++--
.../eventmesh/runtime/demo/AsyncPubClient.java | 5 +-
.../eventmesh/runtime/demo/AsyncSubClient.java | 5 +-
.../eventmesh/runtime/demo/BroadCastSubClient.java | 5 +-
.../apache/eventmesh/runtime/demo/CCSubClient.java | 5 +-
.../apache/eventmesh/runtime/demo/CClientDemo.java | 5 +-
.../eventmesh/runtime/demo/SyncPubClient.java | 7 +-
.../eventmesh/runtime/demo/SyncSubClient.java | 5 +-
.../client/catalog/EventMeshCatalogClient.java | 5 +-
.../grpc/consumer/EventMeshGrpcConsumer.java | 37 ++-----
.../client/grpc/consumer/SubStreamHandler.java | 41 +++-----
.../client/grpc/producer/CloudEventProducer.java | 38 ++-----
.../grpc/producer/EventMeshGrpcProducer.java | 9 +-
.../grpc/producer/EventMeshMessageProducer.java | 25 ++---
.../http/consumer/EventMeshHttpConsumer.java | 14 +--
.../eventmesh/client/http/util/HttpUtils.java | 9 +-
.../client/tcp/common/RequestContext.java | 5 +-
.../eventmesh/client/tcp/common/TcpClient.java | 29 ++----
.../EventMeshMessageTCPPubClient.java | 5 +-
.../client/workflow/EventMeshWorkflowClient.java | 5 +-
.../client/http/demo/SyncRequestInstance.java | 7 +-
.../admin/AdminWebHookConfigOperationManager.java | 6 +-
.../webhook/admin/FileWebHookConfigOperation.java | 35 ++-----
.../webhook/admin/NacosWebHookConfigOperation.java | 15 +--
.../webhook/receive/WebHookController.java | 9 +-
103 files changed, 592 insertions(+), 1129 deletions(-)
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileManager.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileManager.java
index 0d2a84885..d45926817 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileManager.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileManager.java
@@ -17,6 +17,8 @@
package org.apache.eventmesh.common.file;
+import org.apache.eventmesh.common.utils.LogUtils;
+
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -60,14 +62,10 @@ public class WatchFileManager {
return;
}
- if (log.isInfoEnabled()) {
- log.info("[WatchFileManager] start close");
- }
+ LogUtils.info(log, "[WatchFileManager] start close");
for (Map.Entry<String, WatchFileTask> entry : WATCH_FILE_TASK_MAP.entrySet()) {
- if (log.isInfoEnabled()) {
- log.info("[WatchFileManager] start to shutdown : {}", entry.getKey());
- }
+ LogUtils.info(log, "[WatchFileManager] start to shutdown : {}", entry.getKey());
try {
entry.getValue().shutdown();
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileTask.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileTask.java
index 2a1b07bca..ac0654022 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileTask.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/file/WatchFileTask.java
@@ -17,6 +17,8 @@
package org.apache.eventmesh.common.file;
+import org.apache.eventmesh.common.utils.LogUtils;
+
import java.nio.file.FileSystem;
import java.nio.file.FileSystems;
import java.nio.file.Path;
@@ -87,9 +89,7 @@ public class WatchFileTask extends Thread {
for (WatchEvent<?> event : events) {
WatchEvent.Kind<?> kind = event.kind();
if (kind.equals(StandardWatchEventKinds.OVERFLOW)) {
- if (log.isWarnEnabled()) {
- log.warn("[WatchFileTask] file overflow: {}", event.context());
- }
+ LogUtils.warn(log, "[WatchFileTask] file overflow: {}", event.context());
continue;
}
precessWatchEvent(event);
@@ -97,9 +97,7 @@ public class WatchFileTask extends Thread {
} catch (InterruptedException ex) {
boolean interrupted = Thread.interrupted();
if (interrupted) {
- if (log.isDebugEnabled()) {
- log.debug("[WatchFileTask] file watch is interrupted");
- }
+ LogUtils.debug(log, "[WatchFileTask] file watch is interrupted");
}
} catch (Exception ex) {
log.error("[WatchFileTask] an exception occurred during file listening : ", ex);
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java
index ba5a59524..6c9a2fadc 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.protocol.tcp.RedirectInfo;
import org.apache.eventmesh.common.protocol.tcp.Subscription;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
@@ -145,9 +146,7 @@ public class Codec {
}
final byte[] headerData = new byte[headerLength];
in.readBytes(headerData);
- if (log.isDebugEnabled()) {
- log.debug("Decode headerJson={}", deserializeBytes(headerData));
- }
+ LogUtils.debug(log, "Decode headerJson={}", deserializeBytes(headerData));
return JsonUtils.parseObject(headerData, Header.class);
}
@@ -157,9 +156,7 @@ public class Codec {
}
final byte[] bodyData = new byte[bodyLength];
in.readBytes(bodyData);
- if (log.isDebugEnabled()) {
- log.debug("Decode bodyJson={}", deserializeBytes(bodyData));
- }
+ LogUtils.debug(log, "Decode bodyJson={}", deserializeBytes(bodyData));
return deserializeBody(deserializeBytes(bodyData), header);
}
@@ -211,9 +208,7 @@ public class Codec {
case REDIRECT_TO_CLIENT:
return JsonUtils.parseObject(bodyJsonString, RedirectInfo.class);
default:
- if (log.isWarnEnabled()) {
- log.warn("Invalidate TCP command: {}", command);
- }
+ LogUtils.warn(log, "Invalidate TCP command: {}", command);
return null;
}
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnection.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnection.java
index 34dfe2182..1b9cdc11d 100644
--- a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnection.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnection.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.connector.jdbc.connection;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.connector.jdbc.JdbcDriverMetaData;
import org.apache.eventmesh.connector.jdbc.config.JdbcConfig;
@@ -34,11 +35,11 @@ import javax.annotation.concurrent.ThreadSafe;
import lombok.extern.slf4j.Slf4j;
-@Slf4j
/**
* JdbcConnection class representing a JDBC connection.
* Implements the AutoCloseable interface.
*/
+@Slf4j
public class JdbcConnection implements AutoCloseable {
private static final int CONNECTION_VALID_CHECK_TIMEOUT_IN_SEC = 3;
@@ -181,9 +182,7 @@ public class JdbcConnection implements AutoCloseable {
return execute(statement -> {
for (String sqlStatement : sqlStatements) {
if (sqlStatement != null) {
- if (log.isDebugEnabled()) {
- log.debug("Executing '{}'", sqlStatement);
- }
+ LogUtils.debug(log, "Executing '{}'", sqlStatement);
statement.execute(sqlStatement);
}
}
@@ -221,9 +220,7 @@ public class JdbcConnection implements AutoCloseable {
try (Statement statement = conn.createStatement()) {
for (String sqlStatement : sqlStatements) {
- if (log.isDebugEnabled()) {
- log.debug("Executing sql statement: {}", sqlStatement);
- }
+ LogUtils.debug(log, "Executing sql statement: {}", sqlStatement);
statement.execute(sqlStatement);
}
}
@@ -296,9 +293,7 @@ public class JdbcConnection implements AutoCloseable {
public JdbcConnection query(String sql, StatementFactory statementFactory, JdbcResultSetConsumer resultConsumer) throws SQLException {
Connection conn = connection();
try (Statement statement = statementFactory.createStatement(conn)) {
- if (log.isDebugEnabled()) {
- log.debug("Query sql '{}'", sql);
- }
+ LogUtils.debug(log, "Query sql '{}'", sql);
try (ResultSet resultSet = statement.executeQuery(sql)) {
if (resultConsumer != null) {
resultConsumer.accept(resultSet);
@@ -336,9 +331,7 @@ public class JdbcConnection implements AutoCloseable {
public <T> T query(String sql, StatementFactory statementFactory, ResultSetMapper<T> resultSetMapper) throws SQLException {
Connection conn = connection();
try (Statement statement = statementFactory.createStatement(conn)) {
- if (log.isDebugEnabled()) {
- log.debug("Query sql '{}'", sql);
- }
+ LogUtils.debug(log, "Query sql '{}'", sql);
try (ResultSet resultSet = statement.executeQuery(sql)) {
if (resultSetMapper != null) {
return resultSetMapper.map(resultSet);
@@ -381,9 +374,7 @@ public class JdbcConnection implements AutoCloseable {
Connection conn = connection();
try (PreparedStatement preparedStatement = preparedStatementFactory.createPreparedStatement(conn, sql)) {
- if (log.isDebugEnabled()) {
- log.debug("Query sql '{}'", sql);
- }
+ LogUtils.debug(log, "Query sql '{}'", sql);
if (preparedParameters != null) {
for (int index = 0; index < preparedParameters.length; ++index) {
final PreparedParameter preparedParameter = preparedParameters[index];
@@ -436,9 +427,7 @@ public class JdbcConnection implements AutoCloseable {
Connection conn = connection();
try (PreparedStatement preparedStatement = preparedStatementFactory.createPreparedStatement(conn, sql)) {
- if (log.isDebugEnabled()) {
- log.debug("Query sql '{}'", sql);
- }
+ LogUtils.debug(log, "Query sql '{}'", sql);
if (preparedParameters != null) {
for (int index = 0; index < preparedParameters.length; ++index) {
final PreparedParameter preparedParameter = preparedParameters[index];
@@ -579,13 +568,9 @@ public class JdbcConnection implements AutoCloseable {
} else {
url = urlWithPlaceholder;
}
- if (log.isDebugEnabled()) {
- log.debug("URL: {}", url);
- }
+ LogUtils.debug(log, "URL: {}", url);
Connection connection = DriverManager.getConnection(url, config.asProperties());
- if (log.isDebugEnabled()) {
- log.debug("User [{}] Connected to {}", config.getUser(), url);
- }
+ LogUtils.debug(log, "User [{}] Connected to {}", config.getUser(), url);
return connection;
};
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngine.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngine.java
index 7b089dc79..40a999f1a 100644
--- a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngine.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngine.java
@@ -424,9 +424,7 @@ public class MysqlCdcEngine extends AbstractCdcEngine<MysqlAntlr4DdlParser, Mysq
* @param event the event to be handled
*/
protected void handleHeartbeatEvent(MysqlJdbcContext context, Event event) {
- if (log.isDebugEnabled()) {
- log.debug("Replication client handle {}", event.getHeader().getEventType());
- }
+ LogUtils.debug(log, "Replication client handle {}", event.getHeader().getEventType());
}
/**
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlDatabaseDialect.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlDatabaseDialect.java
index 2e76c7410..41436049b 100644
--- a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlDatabaseDialect.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlDatabaseDialect.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.connector.jdbc.source.dialect.mysql;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.connector.jdbc.DataTypeConvertor;
import org.apache.eventmesh.connector.jdbc.JdbcDriverMetaData;
import org.apache.eventmesh.connector.jdbc.connection.mysql.MysqlJdbcConnection;
@@ -137,9 +138,7 @@ public class MysqlDatabaseDialect extends AbstractGeneralDatabaseDialect<MysqlJd
// Build the SQL query to return a list of tables for the given database
String sql = MysqlDialectSql.SHOW_DATABASE_TABLE.ofWrapperSQL("`" + databaseName + "`");
- if (log.isDebugEnabled()) {
- log.debug("List tables SQL:{}", sql);
- }
+ LogUtils.debug(log, "List tables SQL:{}", sql);
this.connection.query(sql, resultSet -> {
// Execute the query and add each table ID to the list
while (resultSet.next()) {
@@ -174,9 +173,7 @@ public class MysqlDatabaseDialect extends AbstractGeneralDatabaseDialect<MysqlJd
// Get table creation SQL
final String createTableSql = MysqlDialectSql.SHOW_CREATE_TABLE.ofWrapperSQL(tableId.getId());
- if (log.isDebugEnabled()) {
- log.debug("Show create table SQL:{}", createTableSql);
- }
+ LogUtils.debug(log, "Show create table SQL:{}", createTableSql);
this.connection.query(createTableSql, resultSet -> {
boolean hasNext = resultSet.next();
@@ -189,9 +186,7 @@ public class MysqlDatabaseDialect extends AbstractGeneralDatabaseDialect<MysqlJd
// Get table columns SQL
final String selectTableSql = MysqlDialectSql.SELECT_TABLE_COLUMNS.ofWrapperSQL(tableId.getId());
- if (log.isDebugEnabled()) {
- log.debug("Select table SQL:{}", selectTableSql);
- }
+ LogUtils.debug(log, "Select table SQL:{}", selectTableSql);
Map<String, DefaultColumn> columns = new HashMap<>(16);
// Execute query to get table columns
this.connection.query(selectTableSql, resultSet -> {
@@ -215,9 +210,7 @@ public class MysqlDatabaseDialect extends AbstractGeneralDatabaseDialect<MysqlJd
// Get table columns details SQL
final String showTableSql = MysqlDialectSql.SHOW_TABLE_COLUMNS.ofWrapperSQL(tableId.getTableName(), tableId.getCatalogName());
- if (log.isDebugEnabled()) {
- log.debug("Show table columns SQL:{}", showTableSql);
- }
+ LogUtils.debug(log, "Show table columns SQL:{}", showTableSql);
// Execute query to get table columns details
List<DefaultColumn> columnList = new ArrayList<>(columns.size());
this.connection.query(showTableSql, resultSet -> {
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/WorkflowAsyncPublishInstance.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/WorkflowAsyncPublishInstance.java
index 5ceb25cf5..6eea77388 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/WorkflowAsyncPublishInstance.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/WorkflowAsyncPublishInstance.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.client.workflow.config.EventMeshWorkflowClientConfig
import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteRequest;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteResponse;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
import org.apache.eventmesh.selector.NacosSelector;
@@ -65,9 +66,7 @@ public class WorkflowAsyncPublishInstance extends GrpcAbstractDemo {
EventMeshWorkflowClient eventMeshWorkflowClient = new EventMeshWorkflowClient(eventMeshWorkflowClientConfig);
ExecuteResponse response = eventMeshWorkflowClient.getWorkflowClient().execute(executeRequest.build());
- if (log.isInfoEnabled()) {
- log.info("received response: {}", response.toString());
- }
+ LogUtils.info(log, "received response: {}", response.toString());
ThreadUtils.sleep(1, TimeUnit.MINUTES);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsAsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsAsyncSubscribe.java
index 32f9baf8c..38c2da59d 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsAsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsAsyncSubscribe.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
@@ -61,9 +62,7 @@ public class CloudEventsAsyncSubscribe extends GrpcAbstractDemo implements Recei
@Override
public Optional<CloudEvent> handle(final CloudEvent msg) {
- if (log.isInfoEnabled()) {
- log.info("receive async msg: {}", msg);
- }
+ LogUtils.info(log, "receive async msg: {}", msg);
return Optional.empty();
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsSubscribeReply.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsSubscribeReply.java
index 53e95c68d..7b529a48a 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsSubscribeReply.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/CloudEventsSubscribeReply.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
@@ -62,9 +63,7 @@ public class CloudEventsSubscribeReply extends GrpcAbstractDemo implements Recei
@Override
public Optional<CloudEvent> handle(final CloudEvent msg) {
- if (log.isInfoEnabled()) {
- log.info("receive request-reply msg: {}", msg);
- }
+ LogUtils.info(log, "receive request-reply msg: {}", msg);
if (msg != null) {
return Optional.of(msg);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshAsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshAsyncSubscribe.java
index b7b7206d3..82a8bef37 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshAsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshAsyncSubscribe.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
@@ -60,9 +61,7 @@ public class EventMeshAsyncSubscribe extends GrpcAbstractDemo implements Receive
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) {
- if (log.isInfoEnabled()) {
- log.info("receive async msg: {}", msg);
- }
+ LogUtils.info(log, "receive async msg: {}", msg);
return Optional.empty();
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeBroadcast.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeBroadcast.java
index 7fcb04cf1..00be1c1ca 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeBroadcast.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeBroadcast.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
@@ -61,9 +62,7 @@ public class EventMeshSubscribeBroadcast extends GrpcAbstractDemo implements Rec
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) {
- if (log.isInfoEnabled()) {
- log.info("receive async broadcast msg: {}", msg);
- }
+ LogUtils.info(log, "receive async broadcast msg: {}", msg);
return Optional.empty();
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeReply.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeReply.java
index 56cab1010..028b5808e 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeReply.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/EventMeshSubscribeReply.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
@@ -60,9 +61,7 @@ public class EventMeshSubscribeReply extends GrpcAbstractDemo implements Receive
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) {
- if (log.isInfoEnabled()) {
- log.info("receive request-reply msg: {}", msg);
- }
+ LogUtils.info(log, "receive request-reply msg: {}", msg);
if (msg != null) {
return Optional.of(msg);
} else {
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowExpressAsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowExpressAsyncSubscribe.java
index 4ba90adca..f3316e80b 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowExpressAsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowExpressAsyncSubscribe.java
@@ -29,6 +29,7 @@ import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteRequest;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteResponse;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
import org.apache.eventmesh.selector.NacosSelector;
@@ -80,9 +81,7 @@ public class WorkflowExpressAsyncSubscribe extends GrpcAbstractDemo implements R
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) throws Exception {
- if (log.isInfoEnabled()) {
- log.info("receive async msg: {}", msg);
- }
+ LogUtils.info(log, "receive async msg: {}", msg);
if (msg == null) {
log.info("async msg is null, workflow end.");
return Optional.empty();
@@ -96,9 +95,7 @@ public class WorkflowExpressAsyncSubscribe extends GrpcAbstractDemo implements R
.setTaskInstanceId(taskInstanceId)
.setInstanceId(workflowInstanceId).build();
final ExecuteResponse response = workflowClient.getWorkflowClient().execute(executeRequest);
- if (log.isInfoEnabled()) {
- log.info("receive workflow msg: {}", response.getInstanceId());
- }
+ LogUtils.info(log, "receive workflow msg: {}", response.getInstanceId());
return Optional.empty();
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowOrderAsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowOrderAsyncSubscribe.java
index 2db100392..e130643ad 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowOrderAsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowOrderAsyncSubscribe.java
@@ -29,6 +29,7 @@ import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteRequest;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteResponse;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
import org.apache.eventmesh.selector.NacosSelector;
@@ -80,9 +81,7 @@ public class WorkflowOrderAsyncSubscribe extends GrpcAbstractDemo implements Rec
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) throws Exception {
- if (log.isInfoEnabled()) {
- log.info("receive async msg: {}", msg);
- }
+ LogUtils.info(log, "receive async msg: {}", msg);
final Map<String, String> props = msg.getProp();
final String workflowInstanceId = props.get("workflowinstanceid");
@@ -92,9 +91,7 @@ public class WorkflowOrderAsyncSubscribe extends GrpcAbstractDemo implements Rec
.setTaskInstanceId(taskInstanceId)
.setInstanceId(workflowInstanceId).build();
final ExecuteResponse response = workflowClient.getWorkflowClient().execute(executeRequest);
- if (log.isInfoEnabled()) {
- log.info("receive workflow msg: {}", response.getInstanceId());
- }
+ LogUtils.info(log, "receive workflow msg: {}", response.getInstanceId());
return Optional.empty();
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowPaymentAsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowPaymentAsyncSubscribe.java
index 7c7c9400f..76f30191d 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowPaymentAsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/WorkflowPaymentAsyncSubscribe.java
@@ -29,6 +29,7 @@ import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.enums.EventMeshProtocolType;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteRequest;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteResponse;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.grpc.GrpcAbstractDemo;
import org.apache.eventmesh.selector.NacosSelector;
@@ -80,9 +81,7 @@ public class WorkflowPaymentAsyncSubscribe extends GrpcAbstractDemo implements R
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) throws Exception {
- if (log.isInfoEnabled()) {
- log.info("receive async msg: {}", msg);
- }
+ LogUtils.info(log, "receive async msg: {}", msg);
if (msg == null) {
log.info("async msg is null, workflow end.");
return Optional.empty();
@@ -96,9 +95,7 @@ public class WorkflowPaymentAsyncSubscribe extends GrpcAbstractDemo implements R
.setTaskInstanceId(taskInstanceId)
.setInstanceId(workflowInstanceId).build();
final ExecuteResponse response = workflowClient.getWorkflowClient().execute(executeRequest);
- if (log.isInfoEnabled()) {
- log.info("receive workflow msg: {}", response.getInstanceId());
- }
+ LogUtils.info(log, "receive workflow msg: {}", response.getInstanceId());
return Optional.empty();
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java
index fa9beb158..c1f563cff 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.grpc.sub.app.controller;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.grpc.sub.app.service.SubService;
import java.util.HashMap;
@@ -43,9 +44,7 @@ public class SubController {
@RequestMapping(value = "/test", method = RequestMethod.POST)
public String subTest(final HttpServletRequest request) {
final String content = request.getParameter("content");
- if (log.isInfoEnabled()) {
- log.info("=======receive message======= {}", content);
- }
+ LogUtils.info(log, "=======receive message======= {}", content);
subService.consumeMessage(content);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/service/SubService.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/service/SubService.java
index 68fe873ef..78d492096 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/service/SubService.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/service/SubService.java
@@ -29,6 +29,7 @@ import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.grpc.pub.eventmeshmessage.AsyncPublishInstance;
import org.apache.eventmesh.util.Utils;
@@ -95,15 +96,11 @@ public class SubService implements InitializingBean {
try {
countDownLatch.await();
} catch (InterruptedException e) {
- if (log.isWarnEnabled()) {
- log.warn("exception occurred when countDownLatch.await ", e);
- }
+ LogUtils.warn(log, "exception occurred when countDownLatch.await ", e);
Thread.currentThread().interrupt();
}
- if (log.isInfoEnabled()) {
- log.info("stopThread start....");
- }
+ LogUtils.info(log, "stopThread start....");
});
@@ -112,9 +109,7 @@ public class SubService implements InitializingBean {
@PreDestroy
public void cleanup() {
- if (log.isInfoEnabled()) {
- log.info("start destroy....");
- }
+ LogUtils.info(log, "start destroy....");
try {
eventMeshGrpcConsumer.unsubscribe(Collections.singletonList(subscriptionItem), url);
@@ -127,21 +122,15 @@ public class SubService implements InitializingBean {
log.error("exception occurred when close consumer ", e);
}
- if (log.isInfoEnabled()) {
- log.info("end destroy....");
- }
+ LogUtils.info(log, "end destroy....");
}
/**
* Count the message already consumed
*/
public void consumeMessage(final String msg) {
- if (log.isInfoEnabled()) {
- log.info("consume message: {}", msg);
- }
+ LogUtils.info(log, "consume message: {}", msg);
countDownLatch.countDown();
- if (log.isInfoEnabled()) {
- log.info("remaining number of messages to be consumed: {}", countDownLatch.getCount());
- }
+ LogUtils.info(log, "remaining number of messages to be consumed: {}", countDownLatch.getCount());
}
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/AsyncSyncRequestInstance.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/AsyncSyncRequestInstance.java
index ba6a31aa3..782e0a11a 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/AsyncSyncRequestInstance.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/AsyncSyncRequestInstance.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.client.http.producer.EventMeshHttpProducer;
import org.apache.eventmesh.client.http.producer.RRCallback;
import org.apache.eventmesh.common.EventMeshMessage;
import org.apache.eventmesh.common.ExampleConstants;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.http.demo.HttpAbstractDemo;
@@ -47,17 +48,13 @@ public class AsyncSyncRequestInstance extends HttpAbstractDemo {
@Override
public void onSuccess(final EventMeshMessage o) {
- if (log.isDebugEnabled()) {
- log.debug("sendmsg: {}, return: {}, cost: {} ms", eventMeshMessage.getContent(), o.getContent(),
- System.currentTimeMillis() - startTime);
- }
+ LogUtils.debug(log, "sendmsg: {}, return: {}, cost: {} ms", eventMeshMessage.getContent(), o.getContent(),
+ System.currentTimeMillis() - startTime);
}
@Override
public void onException(Throwable e) {
- if (log.isDebugEnabled()) {
- log.debug("send msg failed", e);
- }
+ LogUtils.debug(log, "send msg failed", e);
}
}, 3_000);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/SyncRequestInstance.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/SyncRequestInstance.java
index fa6487daf..04305fa9c 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/SyncRequestInstance.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/pub/eventmeshmessage/SyncRequestInstance.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.http.demo.pub.eventmeshmessage;
import org.apache.eventmesh.client.http.producer.EventMeshHttpProducer;
import org.apache.eventmesh.common.EventMeshMessage;
import org.apache.eventmesh.common.ExampleConstants;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.http.demo.HttpAbstractDemo;
@@ -47,10 +48,8 @@ public class SyncRequestInstance extends HttpAbstractDemo {
.build();
final EventMeshMessage rsp = eventMeshHttpProducer.request(eventMeshMessage, 10_000);
- if (log.isDebugEnabled()) {
- log.debug("send msg: {}, return: {}, cost:{} ms", eventMeshMessage.getContent(), rsp.getContent(),
- System.currentTimeMillis() - startTime);
- }
+ LogUtils.debug(log, "send msg: {}, return: {}, cost:{} ms", eventMeshMessage.getContent(), rsp.getContent(),
+ System.currentTimeMillis() - startTime);
} catch (Exception e) {
log.error("send msg failed, ", e);
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/controller/SubController.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/controller/SubController.java
index 8bb2bd495..f3b35ba77 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/controller/SubController.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/controller/SubController.java
@@ -21,6 +21,7 @@ import static org.apache.eventmesh.common.Constants.CLOUD_EVENTS_PROTOCOL_NAME;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.http.demo.sub.service.SubService;
import org.apache.commons.lang3.StringUtils;
@@ -55,11 +56,8 @@ public class SubController {
@RequestMapping(value = "/test", method = RequestMethod.POST)
public String subTest(final HttpServletRequest request) {
final String content = request.getParameter("content");
- if (log.isInfoEnabled()) {
- log.info("receive message: {}", content);
- }
- @SuppressWarnings("unchecked")
- final Map<String, String> contentMap = JsonUtils.parseObject(content, HashMap.class);
+ LogUtils.info(log, "receive message: {}", content);
+ @SuppressWarnings("unchecked") final Map<String, String> contentMap = JsonUtils.parseObject(content, HashMap.class);
if (StringUtils.equals(CLOUD_EVENTS_PROTOCOL_NAME, contentMap.get(ProtocolKey.PROTOCOL_TYPE))) {
final EventFormat eventFormat = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);
if (eventFormat != null) {
@@ -67,9 +65,7 @@ public class SubController {
final CloudEventData eventData = event.getData();
if (eventData != null) {
final String data = new String(eventData.toBytes(), StandardCharsets.UTF_8);
- if (log.isInfoEnabled()) {
- log.info("receive data: {}", data);
- }
+ LogUtils.info(log, "receive data: {}", data);
}
}
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/service/SubService.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/service/SubService.java
index 98655704c..5b39974f9 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/service/SubService.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/http/demo/sub/service/SubService.java
@@ -30,6 +30,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.http.demo.pub.eventmeshmessage.AsyncPublishInstance;
import org.apache.eventmesh.util.Utils;
@@ -101,18 +102,14 @@ public class SubService implements InitializingBean {
log.error("interrupted exception", e);
Thread.currentThread().interrupt();
}
- if (log.isInfoEnabled()) {
- log.info("stopThread start....");
- }
+ LogUtils.info(log, "stopThread start....");
});
stopThread.start();
}
@PreDestroy
public void cleanup() {
- if (log.isInfoEnabled()) {
- log.info("start destroy....");
- }
+ LogUtils.info(log, "start destroy....");
try {
final List<String> unSubList = new ArrayList<>();
@@ -128,21 +125,15 @@ public class SubService implements InitializingBean {
eventMeshHttpConsumer.close();
}
- if (log.isInfoEnabled()) {
- log.info("end destroy....");
- }
+ LogUtils.info(log, "end destroy....");
}
/**
* Count the message already consumed
*/
public void consumeMessage(final String msg) {
- if (log.isInfoEnabled()) {
- log.info("consume message: {}", msg);
- }
+ LogUtils.info(log, "consume message: {}", msg);
countDownLatch.countDown();
- if (log.isInfoEnabled()) {
- log.info("remaining number: {} of messages to be consumed", countDownLatch.getCount());
- }
+ LogUtils.info(log, "remaining number: {} of messages to be consumed", countDownLatch.getCount());
}
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/AsyncPublish.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/AsyncPublish.java
index f2cf36a6e..c340223c4 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/AsyncPublish.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/AsyncPublish.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -54,9 +55,7 @@ public class AsyncPublish {
for (int i = 0; i < 2; i++) {
CloudEvent event = EventMeshTestUtils.generateCloudEventV1Async();
- if (log.isInfoEnabled()) {
- log.info("begin send async msg[{}]: {}", i, event);
- }
+ LogUtils.info(log, "begin send async msg[{}]: {}", i, event);
client.publish(event, EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
ThreadUtils.sleep(1, TimeUnit.SECONDS);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/SyncRequest.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/SyncRequest.java
index 5967f42d4..fda113e37 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/SyncRequest.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/SyncRequest.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -60,9 +61,7 @@ public class SyncRequest {
final CloudEvent event = EventMeshTestUtils.generateCloudEventV1SyncRR();
- if (log.isInfoEnabled()) {
- log.info("begin send rr msg: {}", event);
- }
+ LogUtils.info(log, "begin send rr msg: {}", event);
final Package response = client.rr(event, EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
// check-NPE EventFormat
@@ -83,9 +82,7 @@ public class SyncRequest {
}
final String content = new String(cloudEventData.toBytes(), StandardCharsets.UTF_8);
- if (log.isInfoEnabled()) {
- log.info("receive rr reply: {}|{}", response, content);
- }
+ LogUtils.info(log, "receive rr reply: {}|{}", response, content);
} catch (Exception e) {
log.error("SyncRequest failed", e);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java
index a1ad04aba..ebe78e8c6 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublish.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -54,9 +55,7 @@ public class AsyncPublish {
for (int i = 0; i < 5; i++) {
final EventMeshMessage eventMeshMessage = EventMeshTestUtils.generateAsyncEventMqMsg();
- if (log.isInfoEnabled()) {
- log.info("begin send async msg[{}]: {}", i, eventMeshMessage);
- }
+ LogUtils.info(log, "begin send async msg[{}]: {}", i, eventMeshMessage);
client.publish(eventMeshMessage, EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
ThreadUtils.sleep(1, TimeUnit.SECONDS);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublishBroadcast.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublishBroadcast.java
index bf04db8ed..17dc19d6e 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublishBroadcast.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/AsyncPublishBroadcast.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -53,9 +54,7 @@ public class AsyncPublishBroadcast {
final EventMeshMessage eventMeshMessage = EventMeshTestUtils.generateBroadcastMqMsg();
- if (log.isInfoEnabled()) {
- log.info("begin send broadcast msg: {}", eventMeshMessage);
- }
+ LogUtils.info(log, "begin send broadcast msg: {}", eventMeshMessage);
client.broadcast(eventMeshMessage, EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
ThreadUtils.sleep(2, TimeUnit.SECONDS);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/SyncRequest.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/SyncRequest.java
index d1bde38d0..8a1053960 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/SyncRequest.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/eventmeshmessage/SyncRequest.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -52,14 +53,10 @@ public class SyncRequest {
final EventMeshMessage eventMeshMessage = EventMeshTestUtils.generateSyncRRMqMsg();
- if (log.isInfoEnabled()) {
- log.info("begin send rr msg: {}", eventMeshMessage);
- }
+ LogUtils.info(log, "begin send rr msg: {}", eventMeshMessage);
final Package response = client.rr(eventMeshMessage, EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
- if (log.isInfoEnabled()) {
- log.info("receive rr reply: {}", response);
- }
+ LogUtils.info(log, "receive rr reply: {}", response);
} catch (Exception e) {
log.error("SyncRequest failed", e);
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/AsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/AsyncSubscribe.java
index c9d2c71ed..8c3b77d2f 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/AsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/AsyncSubscribe.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -72,9 +73,7 @@ public class AsyncSubscribe implements ReceiveMsgHook<CloudEvent> {
}
final String content = new String(msg.getData().toBytes(), StandardCharsets.UTF_8);
- if (log.isInfoEnabled()) {
- log.info("receive async msg: {}|{}", msg, content);
- }
+ LogUtils.info(log, "receive async msg: {}|{}", msg, content);
return Optional.empty();
}
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/SyncResponse.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/SyncResponse.java
index 854ec27df..43a2997e5 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/SyncResponse.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/cloudevents/SyncResponse.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.ExampleConstants;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -73,9 +74,7 @@ public class SyncResponse implements ReceiveMsgHook<CloudEvent> {
}
final String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
- if (log.isInfoEnabled()) {
- log.info("receive sync rr msg: {}|{}", event, content);
- }
+ LogUtils.info(log, "receive sync rr msg: {}|{}", event, content);
return Optional.of(event);
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java
index 7b7fb4130..b5b801962 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribe.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -65,9 +66,7 @@ public class AsyncSubscribe implements ReceiveMsgHook<EventMeshMessage> {
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) {
- if (log.isInfoEnabled()) {
- log.info("receive async msg: {}", msg);
- }
+ LogUtils.info(log, "receive async msg: {}", msg);
return Optional.empty();
}
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribeBroadcast.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribeBroadcast.java
index f114c59e0..78b1408ea 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribeBroadcast.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/AsyncSubscribeBroadcast.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -67,9 +68,7 @@ public class AsyncSubscribeBroadcast implements ReceiveMsgHook<EventMeshMessage>
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) {
- if (log.isInfoEnabled()) {
- log.info("receive broadcast msg: {}", msg);
- }
+ LogUtils.info(log, "receive broadcast msg: {}", msg);
return Optional.empty();
}
diff --git a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/SyncResponse.java b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/SyncResponse.java
index ccd9829ec..953bec117 100644
--- a/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/SyncResponse.java
+++ b/eventmesh-examples/src/main/java/org/apache/eventmesh/tcp/demo/sub/eventmeshmessage/SyncResponse.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.tcp.common.EventMeshTestUtils;
import org.apache.eventmesh.util.Utils;
@@ -66,9 +67,7 @@ public class SyncResponse implements ReceiveMsgHook<EventMeshMessage> {
@Override
public Optional<EventMeshMessage> handle(final EventMeshMessage msg) {
- if (log.isInfoEnabled()) {
- log.info("receive sync rr msg: {}", msg);
- }
+ LogUtils.info(log, "receive sync rr msg: {}", msg);
return Optional.ofNullable(msg);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java
index 6a5190a23..bfd5ad496 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java
@@ -127,7 +127,7 @@ public class RedirectClientByIpPortHandler extends AbstractHttpHandler {
} catch (Exception e) {
log.error("clientManage|redirectClientByIpPort|fail|ip={}|port={}|destEventMeshIp"
+
- "={}|destEventMeshPort={},errMsg={}", ip, port, destEventMeshIp, destEventMeshPort, e);
+ "={}|destEventMeshPort={}", ip, port, destEventMeshIp, destEventMeshPort, e);
result = String.format("redirectClientByIpPort fail! sessionMap size {%d}, {clientIp=%s clientPort=%s "
+
"destEventMeshIp=%s destEventMeshPort=%s}, result {%s}, errorMsg : %s",
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java
index b38ab84d4..0b9e8f52c 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java
@@ -124,7 +124,7 @@ public class RedirectClientByPathHandler extends AbstractHttpHandler {
} catch (Exception e) {
log.error("clientManage|redirectClientByPath|fail|path={}|destEventMeshIp"
+
- "={}|destEventMeshPort={},errMsg={}", path, destEventMeshIp, destEventMeshPort, e);
+ "={}|destEventMeshPort={}", path, destEventMeshIp, destEventMeshPort, e);
result = String.format("redirectClientByPath fail! sessionMap size {%d}, {path=%s "
+
"destEventMeshIp=%s destEventMeshPort=%s}, result {%s}, errorMsg : %s",
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientBySubSystemHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientBySubSystemHandler.java
index 48f7a483a..0ba316e03 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientBySubSystemHandler.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientBySubSystemHandler.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.admin.handler;
import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.NetUtils;
import org.apache.eventmesh.runtime.admin.controller.HttpHandlerManager;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
@@ -71,8 +72,7 @@ public class RedirectClientBySubSystemHandler extends AbstractHttpHandler {
* @param httpHandlerManager Manages the registration of {@linkplain com.sun.net.httpserver.HttpHandler HttpHandler}
* for an {@link com.sun.net.httpserver.HttpServer HttpServer}.
*/
- public RedirectClientBySubSystemHandler(final EventMeshTCPServer eventMeshTCPServer,
- final HttpHandlerManager httpHandlerManager) {
+ public RedirectClientBySubSystemHandler(final EventMeshTCPServer eventMeshTCPServer, final HttpHandlerManager httpHandlerManager) {
super(httpHandlerManager);
this.eventMeshTCPServer = eventMeshTCPServer;
}
@@ -102,10 +102,8 @@ public class RedirectClientBySubSystemHandler extends AbstractHttpHandler {
out.write("params illegal!".getBytes(Constants.DEFAULT_CHARSET));
return;
}
- if (log.isInfoEnabled()) {
- log.info("redirectClientBySubSystem in admin,subsys:{},destIp:{},destPort:{}====================",
- subSystem, destEventMeshIp, destEventMeshPort);
- }
+ LogUtils.info(log, "redirectClientBySubSystem in admin,subsys:{},destIp:{},destPort:{}====================",
+ subSystem, destEventMeshIp, destEventMeshPort);
// Retrieve the mapping between Sessions and their corresponding client address
final ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
@@ -127,22 +125,18 @@ public class RedirectClientBySubSystemHandler extends AbstractHttpHandler {
}
} catch (Exception e) {
log.error("clientManage|redirectClientBySubSystem|fail|subSystem={}|destEventMeshIp"
- +
- "={}|destEventMeshPort={},errMsg={}", subSystem, destEventMeshIp, destEventMeshPort, e);
+ + "={}|destEventMeshPort={}", subSystem, destEventMeshIp, destEventMeshPort, e);
NetUtils.sendSuccessResponseHeaders(httpExchange);
out.write(String.format("redirectClientBySubSystem fail! sessionMap size {%d}, {subSystem=%s "
- +
- "destEventMeshIp=%s destEventMeshPort=%s}, result {%s}, errorMsg : %s",
- sessionMap.size(), subSystem, destEventMeshIp, destEventMeshPort, redirectResult, e
- .getMessage())
+ + "destEventMeshIp=%s destEventMeshPort=%s}, result {%s}, errorMsg : %s",
+ sessionMap.size(), subSystem, destEventMeshIp, destEventMeshPort, redirectResult, e.getMessage())
.getBytes(Constants.DEFAULT_CHARSET));
return;
}
NetUtils.sendSuccessResponseHeaders(httpExchange);
out.write(String.format("redirectClientBySubSystem success! sessionMap size {%d}, {subSystem=%s "
- +
- "destEventMeshIp=%s destEventMeshPort=%s}, result {%s} ",
+ + "destEventMeshIp=%s destEventMeshPort=%s}, result {%s} ",
sessionMap.size(), subSystem, destEventMeshIp, destEventMeshPort, redirectResult)
.getBytes(Constants.DEFAULT_CHARSET));
} catch (Exception e) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectAllClientHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectAllClientHandler.java
index 989063667..964021dcd 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectAllClientHandler.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectAllClientHandler.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.admin.handler;
import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.NetUtils;
import org.apache.eventmesh.runtime.admin.controller.HttpHandlerManager;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
@@ -82,9 +83,7 @@ public class RejectAllClientHandler extends AbstractHttpHandler {
final ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
final List<InetSocketAddress> successRemoteAddrs = new ArrayList<>();
try {
- if (log.isInfoEnabled()) {
- log.info("rejectAllClient in admin====================");
- }
+ LogUtils.info(log, "rejectAllClient in admin====================");
if (!sessionMap.isEmpty()) {
// Iterate through the sessionMap and close each client connection
for (final Map.Entry<InetSocketAddress, Session> entry : sessionMap.entrySet()) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientByIpPortHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientByIpPortHandler.java
index b19ca6ecf..af2d5d132 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientByIpPortHandler.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientByIpPortHandler.java
@@ -122,7 +122,7 @@ public class RejectClientByIpPortHandler extends AbstractHttpHandler {
}
}
} catch (Exception e) {
- log.error("clientManage|rejectClientByIpPort|fail|ip={}|port={},errMsg={}", ip, port, e);
+ log.error("clientManage|rejectClientByIpPort|fail|ip={}|port={}", ip, port, e);
result = String.format("rejectClientByIpPort fail! {ip=%s port=%s}, had reject {%s}, errorMsg : %s", ip,
port, NetUtils.addressToString(successRemoteAddrs), e.getMessage());
NetUtils.sendSuccessResponseHeaders(httpExchange);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientBySubSystemHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientBySubSystemHandler.java
index f9b45d744..4ef5d882a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientBySubSystemHandler.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientBySubSystemHandler.java
@@ -131,7 +131,7 @@ public class RejectClientBySubSystemHandler extends AbstractHttpHandler {
}
}
} catch (Exception e) {
- log.error("clientManage|rejectClientBySubSystem|fail|subSystemId={},errMsg={}", subSystem, e);
+ log.error("clientManage|rejectClientBySubSystem|fail|subSystemId={}", subSystem, e);
result = String.format("rejectClientBySubSystem fail! sessionMap size {%d}, had reject {%s} , {"
+
"subSystemId=%s}, errorMsg : %s", sessionMap.size(), printClients(successRemoteAddrs),
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientBySystemHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientBySystemHandler.java
index 7e5dbf996..1b65ea4e4 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientBySystemHandler.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientBySystemHandler.java
@@ -19,6 +19,7 @@ package org.apache.eventmesh.runtime.admin.handler;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.NetUtils;
import org.apache.eventmesh.runtime.admin.controller.HttpHandlerManager;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
@@ -84,9 +85,7 @@ public class ShowClientBySystemHandler extends AbstractHttpHandler {
String subSystem = queryStringInfo.get(EventMeshConstants.MANAGE_SUBSYSTEM);
String newLine = System.getProperty("line.separator");
- if (log.isInfoEnabled()) {
- log.info("showClientBySubsys,subsys:{}", subSystem);
- }
+ LogUtils.info(log, "showClientBySubsys,subsys:{}", subSystem);
// Retrieve the mapping between Sessions and their corresponding client address
ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
index 37cb3d050..a8d9e41e6 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.http.header.Header;
import org.apache.eventmesh.common.utils.AssertUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.common.Pair;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -161,9 +162,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
.childHandler(new HttpsServerInitializer(useTLS ? SSLContextFactory.getSslContext(eventMeshHttpConfiguration) : null))
.childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
- if (log.isInfoEnabled()) {
- log.info("HTTPServer[port={}] started.", this.getPort());
- }
+ LogUtils.info(log, "HTTPServer[port={}] started.", this.getPort());
bootstrap.bind(this.getPort())
.channel()
@@ -241,10 +240,8 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
public void sendResponse(final ChannelHandlerContext ctx, final DefaultFullHttpResponse response) {
ctx.writeAndFlush(response).addListener((ChannelFutureListener) f -> {
if (!f.isSuccess()) {
- if (log.isWarnEnabled()) {
- log.warn("send response to [{}] fail, will close this channel",
- RemotingHelper.parseChannelRemoteAddr(f.channel()));
- }
+ LogUtils.warn(log, "send response to [{}] fail, will close this channel",
+ RemotingHelper.parseChannelRemoteAddr(f.channel()));
f.channel().close();
}
});
@@ -363,9 +360,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
return;
}
- if (log.isDebugEnabled()) {
- log.debug("{}", requestCommand);
- }
+ LogUtils.debug(log, "{}", requestCommand);
final AsyncContext<HttpCommand> asyncContext =
new AsyncContext<>(requestCommand, responseCommand, asyncContextCompleteHandler);
@@ -413,9 +408,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
if (asyncContext.isComplete()) {
sendResponse(ctx, responseCommand.httpResponse());
- if (log.isDebugEnabled()) {
- log.debug("{}", asyncContext.getResponse());
- }
+ LogUtils.debug(log, "{}", asyncContext.getResponse());
final Map<String, Object> traceMap = asyncContext.getRequest().getHeader().toMap();
TraceUtils.finishSpanWithException(TraceUtils.prepareServerSpan(traceMap,
EventMeshTraceConstants.TRACE_UPSTREAM_EVENTMESH_SERVER_SPAN,
@@ -435,9 +428,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
metrics.getSummaryMetrics()
.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
- if (log.isDebugEnabled()) {
- log.debug("{}", asyncContext.getResponse());
- }
+ LogUtils.debug(log, "{}", asyncContext.getResponse());
sendResponse(ctx, asyncContext.getResponse().httpResponse());
@@ -493,10 +484,8 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
if (connections.incrementAndGet() > MAX_CONNECTIONS) {
- if (log.isWarnEnabled()) {
- log.warn("client|http|channelActive|remoteAddress={}|msg=too many client({}) connect this eventMesh server",
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()), MAX_CONNECTIONS);
- }
+ LogUtils.warn(log, "client|http|channelActive|remoteAddress={}|msg=too many client({}) connect this eventMesh server",
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()), MAX_CONNECTIONS);
ctx.close();
return;
}
@@ -515,10 +504,8 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
final IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.ALL_IDLE)) {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- if (log.isInfoEnabled()) {
- log.info("client|http|userEventTriggered|remoteAddress={}|msg={}",
- remoteAddress, evt.getClass().getName());
- }
+ LogUtils.info(log, "client|http|userEventTriggered|remoteAddress={}|msg={}",
+ remoteAddress, evt.getClass().getName());
ctx.close();
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
index 538c8ca17..0a876c413 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.boot;
import org.apache.eventmesh.common.EventMeshThreadFactory;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.SystemUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
@@ -78,25 +79,20 @@ public abstract class AbstractRemotingServer {
public void shutdown() throws Exception {
if (bossGroup != null) {
bossGroup.shutdownGracefully();
- if (log.isInfoEnabled()) {
- log.info("shutdown bossGroup");
- }
+ LogUtils.info(log, "shutdown bossGroup");
}
ThreadUtils.randomPause(TimeUnit.SECONDS.toMillis(DEFAULT_SLEEP_SECONDS));
if (ioGroup != null) {
ioGroup.shutdownGracefully();
- if (log.isInfoEnabled()) {
- log.info("shutdown ioGroup");
- }
+ LogUtils.info(log, "shutdown ioGroup");
}
if (workerGroup != null) {
workerGroup.shutdownGracefully();
- if (log.isInfoEnabled()) {
- log.info("shutdown workerGroup");
- }
+
+ LogUtils.info(log, "shutdown workerGroup");
}
}
@@ -108,30 +104,30 @@ public abstract class AbstractRemotingServer {
return bossGroup;
}
- public EventLoopGroup getIoGroup() {
- return ioGroup;
- }
-
- public EventExecutorGroup getWorkerGroup() {
- return workerGroup;
- }
-
- public int getPort() {
- return port;
- }
-
public void setBossGroup(final EventLoopGroup bossGroup) {
this.bossGroup = bossGroup;
}
+ public EventLoopGroup getIoGroup() {
+ return ioGroup;
+ }
+
public void setIoGroup(final EventLoopGroup ioGroup) {
this.ioGroup = ioGroup;
}
+ public EventExecutorGroup getWorkerGroup() {
+ return workerGroup;
+ }
+
public void setWorkerGroup(final EventExecutorGroup workerGroup) {
this.workerGroup = workerGroup;
}
+ public int getPort() {
+ return port;
+ }
+
public void setPort(final int port) {
this.port = port;
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractTCPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractTCPServer.java
index 3fcce8ebf..911e87f4b 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractTCPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractTCPServer.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
import org.apache.eventmesh.common.protocol.tcp.codec.Codec;
import org.apache.eventmesh.common.utils.AssertUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.common.Pair;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -247,17 +248,13 @@ public class AbstractTCPServer extends AbstractRemotingServer {
}
if (Command.HELLO_REQUEST == cmd || Command.RECOMMEND_REQUEST == cmd) {
- if (messageLogger.isInfoEnabled()) {
- messageLogger.info("pkg|c2eventMesh|cmd={}|pkg={}", cmd, pkg);
- }
+ LogUtils.info(messageLogger, "pkg|c2eventMesh|cmd={}|pkg={}", cmd, pkg);
processHttpCommandRequest(pkg, ctx, startTime, cmd);
return;
}
if (clientSessionGroupMapping.getSession(ctx) == null) {
- if (messageLogger.isInfoEnabled()) {
- messageLogger.info("pkg|c2eventMesh|cmd={}|pkg={},no session is found", cmd, pkg);
- }
+ LogUtils.info(messageLogger, "pkg|c2eventMesh|cmd={}|pkg={},no session is found", cmd, pkg);
throw new Exception("no session is found");
}
@@ -300,8 +297,8 @@ public class AbstractTCPServer extends AbstractRemotingServer {
private boolean isNeedTrace(Command cmd) {
return eventMeshTCPConfiguration.isEventMeshServerTraceEnable()
&& (Command.REQUEST_TO_SERVER == cmd
- || Command.ASYNC_MESSAGE_TO_SERVER == cmd
- || Command.BROADCAST_MESSAGE_TO_SERVER == cmd);
+ || Command.ASYNC_MESSAGE_TO_SERVER == cmd
+ || Command.BROADCAST_MESSAGE_TO_SERVER == cmd);
}
private void writeToClient(Command cmd, Package pkg, ChannelHandlerContext ctx, Exception e) {
@@ -375,7 +372,7 @@ public class AbstractTCPServer extends AbstractRemotingServer {
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
Session session = clientSessionGroupMapping.getSession(ctx);
UserAgent client = session == null ? null : session.getClient();
- log.error("exceptionCaught, push goodbye to client|user={},errMsg={}", client, cause.fillInStackTrace());
+ log.error("exceptionCaught, push goodbye to client|user={}, errMsg={}", client, cause.fillInStackTrace());
String errMsg;
if (cause.toString().contains("value not one of declared Enum instance names")) {
errMsg = "Unknown Command type";
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index 508520eab..2cfe2d28d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.api.meta.dto.EventMeshUnRegisterInfo;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.MetricsPluginFactory;
import org.apache.eventmesh.metrics.api.MetricsRegistry;
import org.apache.eventmesh.runtime.acl.Acl;
@@ -73,7 +74,6 @@ import lombok.extern.slf4j.Slf4j;
/**
* Add multiple managers to the underlying server
- *
*/
@Slf4j
public class EventMeshHTTPServer extends AbstractHTTPServer {
@@ -109,9 +109,7 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
}
public void init() throws Exception {
- if (log.isInfoEnabled()) {
- log.info("==================EventMeshHTTPServer Initialing==================");
- }
+ LogUtils.info(log, "==================EventMeshHTTPServer Initialing==================");
super.init();
msgRateLimiter = RateLimiter.create(eventMeshHttpConfiguration.getEventMeshHttpMsgReqNumPerSecond());
@@ -146,9 +144,7 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
registerHTTPRequestProcessor();
- if (log.isInfoEnabled()) {
- log.info("==================EventMeshHTTPServer initialized==================");
- }
+ LogUtils.info(log, "==================EventMeshHTTPServer initialized==================");
}
@Override
@@ -163,9 +159,7 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
if (eventMeshHttpConfiguration.isEventMeshServerMetaStorageEnable()) {
this.register();
}
- if (log.isInfoEnabled()) {
- log.info("==================EventMeshHTTPServer started==================");
- }
+ LogUtils.info(log, "==================EventMeshHTTPServer started==================");
}
@Override
@@ -186,9 +180,7 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
if (eventMeshHttpConfiguration.isEventMeshServerMetaStorageEnable()) {
this.unRegister();
}
- if (log.isInfoEnabled()) {
- log.info("==================EventMeshHTTPServer shutdown==================");
- }
+ LogUtils.info(log, "==================EventMeshHTTPServer shutdown==================");
}
/**
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
index 832cc75c8..9f5cb82b5 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.config.CommonConfiguration;
import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.utils.AssertUtils;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.admin.controller.ClientManageController;
import org.apache.eventmesh.runtime.common.ServiceState;
@@ -141,16 +142,12 @@ public class EventMeshServer {
final String eventStore = System.getProperty(EventMeshConstants.EVENT_STORE_PROPERTIES, System.getenv(EventMeshConstants.EVENT_STORE_ENV));
- if (log.isInfoEnabled()) {
- log.info("eventStore : {}", eventStore);
- }
+ LogUtils.info(log, "eventStore : {}", eventStore);
producerTopicManager = new ProducerTopicManager(this);
producerTopicManager.init();
serviceState = ServiceState.INITED;
- if (log.isInfoEnabled()) {
- log.info(SERVER_STATE_MSG, serviceState);
- }
+ LogUtils.info(log, SERVER_STATE_MSG, serviceState);
}
public void start() throws Exception {
@@ -173,17 +170,13 @@ public class EventMeshServer {
}
producerTopicManager.start();
serviceState = ServiceState.RUNNING;
- if (log.isInfoEnabled()) {
- log.info(SERVER_STATE_MSG, serviceState);
- }
+ LogUtils.info(log, SERVER_STATE_MSG, serviceState);
}
public void shutdown() throws Exception {
serviceState = ServiceState.STOPPING;
- if (log.isInfoEnabled()) {
- log.info(SERVER_STATE_MSG, serviceState);
- }
+ LogUtils.info(log, SERVER_STATE_MSG, serviceState);
for (final EventMeshBootstrap eventMeshBootstrap : BOOTSTRAP_LIST) {
eventMeshBootstrap.shutdown();
@@ -206,9 +199,7 @@ public class EventMeshServer {
ConfigurationContextUtil.clear();
serviceState = ServiceState.STOPPED;
- if (log.isInfoEnabled()) {
- log.info(SERVER_STATE_MSG, serviceState);
- }
+ LogUtils.info(log, SERVER_STATE_MSG, serviceState);
}
public static Trace getTrace() {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java
index c5b3220db..2a00634b0 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.boot;
import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.util.BannerUtil;
@@ -40,15 +41,12 @@ public class EventMeshStartup {
server.start();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
try {
- if (log.isInfoEnabled()) {
- log.info("eventMesh shutting down hook begin.");
- }
+ LogUtils.info(log, "eventMesh shutting down hook begin.");
long start = System.currentTimeMillis();
server.shutdown();
long end = System.currentTimeMillis();
- if (log.isInfoEnabled()) {
- log.info("eventMesh shutdown cost {}ms", end - start);
- }
+
+ LogUtils.info(log, "eventMesh shutdown cost {}ms", end - start);
} catch (Exception e) {
log.error("exception when shutdown.", e);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
index ae289799f..548e891c8 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.api.meta.dto.EventMeshUnRegisterInfo;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.metrics.api.MetricsPluginFactory;
import org.apache.eventmesh.metrics.api.MetricsRegistry;
@@ -60,7 +61,6 @@ import lombok.extern.slf4j.Slf4j;
/**
* Add multiple managers to the underlying server
- *
*/
@Slf4j
public class EventMeshTCPServer extends AbstractTCPServer {
@@ -89,9 +89,7 @@ public class EventMeshTCPServer extends AbstractTCPServer {
}
public void init() throws Exception {
- if (log.isInfoEnabled()) {
- log.info("==================EventMeshTCPServer Initialing==================");
- }
+ LogUtils.info(log, "==================EventMeshTCPServer Initialing==================");
super.init();
rateLimiter = RateLimiter.create(eventMeshTCPConfiguration.getEventMeshTcpMsgReqnumPerSecond());
@@ -121,9 +119,7 @@ public class EventMeshTCPServer extends AbstractTCPServer {
registerTCPRequestProcessor();
- if (log.isInfoEnabled()) {
- log.info("--------------------------EventMeshTCPServer Inited");
- }
+ LogUtils.info(log, "--------------------------EventMeshTCPServer Inited");
}
@Override
@@ -139,9 +135,7 @@ public class EventMeshTCPServer extends AbstractTCPServer {
eventMeshRebalanceService.start();
}
- if (log.isInfoEnabled()) {
- log.info("--------------------------EventMeshTCPServer Started");
- }
+ LogUtils.info(log, "--------------------------EventMeshTCPServer Started");
}
@Override
@@ -160,9 +154,7 @@ public class EventMeshTCPServer extends AbstractTCPServer {
this.unRegister();
}
- if (log.isInfoEnabled()) {
- log.info("--------------------------EventMeshTCPServer Shutdown");
- }
+ LogUtils.info(log, "--------------------------EventMeshTCPServer Shutdown");
}
/**
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java
index b79d3ec33..e4ec6a7a6 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java
@@ -19,6 +19,7 @@ package org.apache.eventmesh.runtime.core.protocol.grpc.consumer;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.common.ServiceState;
import org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupClient;
@@ -64,16 +65,12 @@ public class ConsumerManager {
}
public void init() throws Exception {
- if (log.isInfoEnabled()) {
- log.info("Grpc ConsumerManager initialized.");
- }
+ LogUtils.info(log, "Grpc ConsumerManager initialized.");
}
public void start() throws Exception {
startClientCheck();
- if (log.isInfoEnabled()) {
- log.info("Grpc ConsumerManager started.");
- }
+ LogUtils.info(log, "Grpc ConsumerManager started.");
}
public void shutdown() throws Exception {
@@ -81,9 +78,7 @@ public class ConsumerManager {
consumer.shutdown();
}
scheduledExecutorService.shutdown();
- if (log.isInfoEnabled()) {
- log.info("Grpc ConsumerManager shutdown.");
- }
+ LogUtils.info(log, "Grpc ConsumerManager shutdown.");
}
public EventMeshConsumer getEventMeshConsumer(final String consumerGroup) {
@@ -203,16 +198,12 @@ public class ConsumerManager {
final int clientTimeout = eventMeshGrpcServer.getEventMeshGrpcConfiguration().getEventMeshSessionExpiredInMills();
if (clientTimeout > 0) {
scheduledExecutorService.scheduleAtFixedRate(() -> {
- if (log.isDebugEnabled()) {
- log.debug("grpc client info check");
- }
+ LogUtils.debug(log, "grpc client info check");
final List<ConsumerGroupClient> clientList = new ArrayList<>();
clientTable.values().forEach(clientList::addAll);
- if (log.isDebugEnabled()) {
- log.debug("total number of ConsumerGroupClients: {}", clientList.size());
- }
+ LogUtils.debug(log, "total number of ConsumerGroupClients: {}", clientList.size());
if (CollectionUtils.isEmpty(clientList)) {
return;
@@ -221,10 +212,8 @@ public class ConsumerManager {
final Set<String> consumerGroupRestart = new HashSet<>();
clientList.forEach(client -> {
if (System.currentTimeMillis() - client.getLastUpTime().getTime() > clientTimeout) {
- if (log.isWarnEnabled()) {
- log.warn("client {} lastUpdate time {} over three heartbeat cycles. Removing it",
- JsonUtils.toJSONString(client), client.getLastUpTime());
- }
+ LogUtils.warn(log, "client {} lastUpdate time {} over three heartbeat cycles. Removing it",
+ JsonUtils.toJSONString(client), client.getLastUpTime());
deregisterClient(client);
if (getEventMeshConsumer(client.getConsumerGroup()).deregisterClient(client)) {
@@ -238,9 +227,7 @@ public class ConsumerManager {
try {
restartEventMeshConsumer(consumerGroup);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("Error in restarting EventMeshConsumer [{}]", consumerGroup, e);
- }
+ LogUtils.error(log, "Error in restarting EventMeshConsumer [{}]", consumerGroup, e);
}
});
}, 10_000, 10_000, TimeUnit.MILLISECONDS);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
index 2ddaa9983..c8567a9e6 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
@@ -34,6 +34,7 @@ import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.common.ServiceState;
@@ -219,9 +220,7 @@ public class EventMeshConsumer {
broadcastMqConsumer.registerEventListener(createEventListener(SubscriptionMode.BROADCASTING));
serviceState = ServiceState.INITED;
- if (log.isInfoEnabled()) {
- log.info("EventMeshConsumer [{}] initialized.............", consumerGroup);
- }
+ LogUtils.info(log, "EventMeshConsumer [{}] initialized.............", consumerGroup);
}
public synchronized void start() throws Exception {
@@ -242,9 +241,7 @@ public class EventMeshConsumer {
broadcastMqConsumer.start();
serviceState = ServiceState.RUNNING;
- if (log.isInfoEnabled()) {
- log.info("EventMeshConsumer [{}] started..........", consumerGroup);
- }
+ LogUtils.info(log, "EventMeshConsumer [{}] started..........", consumerGroup);
}
public synchronized void shutdown() throws Exception {
@@ -252,9 +249,7 @@ public class EventMeshConsumer {
broadcastMqConsumer.shutdown();
serviceState = ServiceState.STOPPED;
- if (log.isInfoEnabled()) {
- log.info("EventMeshConsumer [{}] shutdown.........", consumerGroup);
- }
+ LogUtils.info(log, "EventMeshConsumer [{}] shutdown.........", consumerGroup);
}
public ServiceState getStatus() {
@@ -352,9 +347,7 @@ public class EventMeshConsumer {
}
}
} else {
- if (log.isDebugEnabled()) {
- log.debug("no active consumer for topic={}|msg={}", topic, event);
- }
+ LogUtils.debug(log, "no active consumer for topic={}|msg={}", topic, event);
}
eventMeshAsyncConsumeContext.commit(EventMeshAction.CommitMessage);
@@ -366,10 +359,8 @@ public class EventMeshConsumer {
final EventMeshProducer producer = eventMeshGrpcServer.getProducerManager().getEventMeshProducer(consumerGroup);
if (producer == null) {
- if (log.isWarnEnabled()) {
- log.warn("consumer:{} consume fail, sendMessageBack, bizSeqNo:{}, uniqueId:{}",
- consumerGroup, bizSeqNo, uniqueId);
- }
+ LogUtils.warn(log, "consumer:{} consume fail, sendMessageBack, bizSeqNo:{}, uniqueId:{}",
+ consumerGroup, bizSeqNo, uniqueId);
return;
}
@@ -384,10 +375,8 @@ public class EventMeshConsumer {
@Override
public void onException(final OnExceptionContext context) {
- if (log.isWarnEnabled()) {
- log.warn("consumer:{} consume fail, sendMessageBack, bizSeqNo:{}, uniqueId:{}", consumerGroup,
- bizSeqNo, uniqueId);
- }
+ LogUtils.warn(log, "consumer:{} consume fail, sendMessageBack, bizSeqNo:{}, uniqueId:{}", consumerGroup,
+ bizSeqNo, uniqueId);
}
});
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
index 8a40cba66..6094e2210 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
@@ -19,6 +19,7 @@ package org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.core.protocol.grpc.service.EventEmitter;
import org.apache.commons.collections4.MapUtils;
@@ -61,10 +62,8 @@ public class StreamTopicConfig extends ConsumerGroupTopicConfig {
Objects.requireNonNull(client, "ConsumerGroupClient can not be null");
if (client.getGrpcType() != grpcType) {
- if (log.isWarnEnabled()) {
- log.warn("Invalid grpc type: {}, expecting grpc type: {}, can not register client {}",
- client.getGrpcType(), grpcType, client);
- }
+ LogUtils.warn(log, "Invalid grpc type: {}, expecting grpc type: {}, can not register client {}",
+ client.getGrpcType(), grpcType, client);
return;
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java
index daee0e206..bd0b0f6c4 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.core.protocol.grpc.consumer.ConsumerManager;
@@ -45,10 +46,10 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class SubscribeProcessor {
- private final EventMeshGrpcServer eventMeshGrpcServer;
-
private static final GrpcType grpcType = GrpcType.WEBHOOK;
+ private final EventMeshGrpcServer eventMeshGrpcServer;
+
private final Acl acl;
public SubscribeProcessor(final EventMeshGrpcServer eventMeshGrpcServer) {
@@ -72,9 +73,7 @@ public class SubscribeProcessor {
try {
doAclCheck(subscription);
} catch (AclException e) {
- if (log.isWarnEnabled()) {
- log.warn("CLIENT HAS NO PERMISSION to Subscribe. failed", e);
- }
+ LogUtils.warn(log, "CLIENT HAS NO PERMISSION to Subscribe. failed", e);
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_ACL_ERR, e.getMessage(), emitter);
return;
}
@@ -127,14 +126,10 @@ public class SubscribeProcessor {
// restart consumer group if required
if (requireRestart) {
- if (log.isInfoEnabled()) {
- log.info("ConsumerGroup {} topic info changed, restart EventMesh Consumer", consumerGroup);
- }
+ LogUtils.info(log, "ConsumerGroup {} topic info changed, restart EventMesh Consumer", consumerGroup);
consumerManager.restartEventMeshConsumer(consumerGroup);
} else {
- if (log.isWarnEnabled()) {
- log.warn("EventMesh consumer [{}] didn't restart.", consumerGroup);
- }
+ LogUtils.warn(log, "EventMesh consumer [{}] didn't restart.", consumerGroup);
}
ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS, "subscribe success", emitter);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/HttpClientGroupMapping.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/HttpClientGroupMapping.java
index ae5400d5e..34aca9907 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/HttpClientGroupMapping.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/HttpClientGroupMapping.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.http.header.client.SubscribeRequestHeader;
import org.apache.eventmesh.common.protocol.http.header.client.UnSubscribeRequestHeader;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupMetadata;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
@@ -254,15 +255,11 @@ public final class HttpClientGroupMapping {
final ConsumerGroupTopicConf currentTopicConf = map.get(subTopic.getTopic());
if (!currentTopicConf.getUrls().add(url)) {
isChange = true;
- if (log.isInfoEnabled()) {
- log.info("add subscribe success, group:{}, url:{} , topic:{}", consumerGroup, url,
- subTopic.getTopic());
- }
+ LogUtils.info(log, "add subscribe success, group:{}, url:{} , topic:{}", consumerGroup, url,
+ subTopic.getTopic());
} else {
- if (log.isWarnEnabled()) {
- log.warn("The group has subscribed, group:{}, url:{} , topic:{}", consumerGroup, url,
- subTopic.getTopic());
- }
+ LogUtils.warn(log, "The group has subscribed, group:{}, url:{} , topic:{}", consumerGroup, url,
+ subTopic.getTopic());
}
if (!currentTopicConf.getIdcUrls().containsKey(clientIdc)) {
@@ -270,24 +267,18 @@ public final class HttpClientGroupMapping {
urls.add(url);
currentTopicConf.getIdcUrls().put(clientIdc, urls);
isChange = true;
- if (log.isInfoEnabled()) {
- log.info("add url to idcUrlMap success, group:{}, url:{}, topic:{}, clientIdc:{}",
- consumerGroup, url, subTopic.getTopic(), clientIdc);
- }
+ LogUtils.info(log, "add url to idcUrlMap success, group:{}, url:{}, topic:{}, clientIdc:{}",
+ consumerGroup, url, subTopic.getTopic(), clientIdc);
} else {
final Set<String> tmpSet = new HashSet<>(currentTopicConf.getIdcUrls().get(clientIdc));
if (!tmpSet.contains(url)) {
currentTopicConf.getIdcUrls().get(clientIdc).add(url);
isChange = true;
- if (log.isInfoEnabled()) {
- log.info("add url to idcUrlMap success, group:{}, url:{}, topic:{}, clientIdc:{}",
- consumerGroup, url, subTopic.getTopic(), clientIdc);
- }
+ LogUtils.info(log, "add url to idcUrlMap success, group:{}, url:{}, topic:{}, clientIdc:{}",
+ consumerGroup, url, subTopic.getTopic(), clientIdc);
} else {
- if (log.isWarnEnabled()) {
- log.warn("The idcUrlMap has contains url, group:{}, url:{} , topic:{}, clientIdc:{}",
- consumerGroup, url, subTopic.getTopic(), clientIdc);
- }
+ LogUtils.warn(log, "The idcUrlMap has contains url, group:{}, url:{} , topic:{}, clientIdc:{}",
+ consumerGroup, url, subTopic.getTopic(), clientIdc);
}
}
}
@@ -306,69 +297,48 @@ public final class HttpClientGroupMapping {
final ConsumerGroupConf consumerGroupConf = localConsumerGroupMapping.get(consumerGroup);
if (consumerGroupConf == null) {
- if (log.isWarnEnabled()) {
- log.warn("unsubscribe fail, the current mesh does not have group subscriptionInfo, group:{}, url:{}",
- consumerGroup, unSubscribeUrl);
- }
+ LogUtils.warn(log, "unsubscribe fail, the current mesh does not have group subscriptionInfo, group:{}, url:{}",
+ consumerGroup, unSubscribeUrl);
return false;
}
final ConsumerGroupTopicConf consumerGroupTopicConf = consumerGroupConf.getConsumerGroupTopicConf().get(unSubTopic);
if (consumerGroupTopicConf == null) {
- if (log.isWarnEnabled()) {
- log.warn(
- "unsubscribe fail, the current mesh does not have group-topic subscriptionInfo, group:{}, topic:{}, url:{}",
- consumerGroup, unSubTopic, unSubscribeUrl);
- }
+ LogUtils.warn(log, "unsubscribe fail, the current mesh does not have group-topic subscriptionInfo, group:{}, topic:{}, url:{}",
+ consumerGroup, unSubTopic, unSubscribeUrl);
return false;
}
if (consumerGroupTopicConf.getUrls().remove(unSubscribeUrl)) {
isChange = true;
- if (log.isInfoEnabled()) {
- log.info("remove url success, group:{}, topic:{}, url:{}", consumerGroup, unSubTopic, unSubscribeUrl);
- }
+ LogUtils.info(log, "remove url success, group:{}, topic:{}, url:{}", consumerGroup, unSubTopic, unSubscribeUrl);
} else {
- if (log.isWarnEnabled()) {
- log.warn("remove url fail, not exist subscrition of this url, group:{}, topic:{}, url:{}",
- consumerGroup, unSubTopic, unSubscribeUrl);
- }
+ LogUtils.warn(log, "remove url fail, not exist subscrition of this url, group:{}, topic:{}, url:{}",
+ consumerGroup, unSubTopic, unSubscribeUrl);
}
if (consumerGroupTopicConf.getIdcUrls().containsKey(clientIdc)) {
if (consumerGroupTopicConf.getIdcUrls().get(clientIdc).remove(unSubscribeUrl)) {
isChange = true;
- if (log.isInfoEnabled()) {
- log.info("remove url from idcUrlMap success, group:{}, topic:{}, url:{}, clientIdc:{}",
- consumerGroup, unSubTopic, unSubscribeUrl, clientIdc);
- }
+ LogUtils.info(log, "remove url from idcUrlMap success, group:{}, topic:{}, url:{}, clientIdc:{}",
+ consumerGroup, unSubTopic, unSubscribeUrl, clientIdc);
} else {
- if (log.isWarnEnabled()) {
- log.warn(
- "remove url from idcUrlMap fail, not exist subscriber of this url, group:{}, topic:{}, url:{}, clientIdc:{}",
- consumerGroup, unSubTopic, unSubscribeUrl, clientIdc);
- }
- }
- } else {
- if (log.isWarnEnabled()) {
- log.warn(
- "remove url from idcUrlMap fail,not exist subscrition of this idc , group:{}, topic:{}, url:{}, clientIdc:{}",
+ LogUtils.warn(log, "remove url from idcUrlMap fail, not exist subscriber of this url, group:{}, topic:{}, url:{}, clientIdc:{}",
consumerGroup, unSubTopic, unSubscribeUrl, clientIdc);
}
+ } else {
+ LogUtils.warn(log, "remove url from idcUrlMap fail,not exist subscrition of this idc , group:{}, topic:{}, url:{}, clientIdc:{}",
+ consumerGroup, unSubTopic, unSubscribeUrl, clientIdc);
}
if (isChange && CollectionUtils.isEmpty(consumerGroupTopicConf.getUrls())) {
consumerGroupConf.getConsumerGroupTopicConf().remove(unSubTopic);
- if (log.isInfoEnabled()) {
- log.info("group unsubscribe topic success,group:{}, topic:{}", consumerGroup, unSubTopic);
- }
+ LogUtils.info(log, "group unsubscribe topic success,group:{}, topic:{}", consumerGroup, unSubTopic);
}
if (isChange && MapUtils.isEmpty(consumerGroupConf.getConsumerGroupTopicConf())) {
localConsumerGroupMapping.remove(consumerGroup);
- if (log.isInfoEnabled()) {
- log.info("group unsubscribe success,group:{}", consumerGroup);
- }
+ LogUtils.info(log, "group unsubscribe success,group:{}", consumerGroup);
}
return isChange;
}
@@ -422,9 +392,9 @@ public final class HttpClientGroupMapping {
for (final String unSubTopic : unSubTopicList) {
isChange = isChange
|| removeSubscriptionByTopic(consumerGroup,
- unSubscribeUrl,
- unSubscribeRequestHeader.getIdc(),
- unSubTopic);
+ unSubscribeUrl,
+ unSubscribeRequestHeader.getIdc(),
+ unSubTopic);
}
} finally {
READ_WRITE_LOCK.writeLock().unlock();
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
index 2f69c99ad..9fc1c0b4a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
@@ -32,6 +32,7 @@ import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchRequestHeader;
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -223,9 +224,7 @@ public class BatchSendMessageProcessor implements HttpRequestProcessor {
topicBatchMessageMappings.put(cloudEvent.getSubject(), tmp);
}
- if (batchMessageLogger.isDebugEnabled()) {
- batchMessageLogger.debug("msg2MQMsg suc, event:{}", cloudEvent.getData());
- }
+ LogUtils.debug(batchMessageLogger, "msg2MQMsg suc, event:{}", cloudEvent.getData());
} catch (Exception e) {
batchMessageLogger.error("msg2MQMsg err, event:{}", cloudEvent.getData(), e);
}
@@ -284,10 +283,9 @@ public class BatchSendMessageProcessor implements HttpRequestProcessor {
long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
summaryMetrics.recordBatchSendMsgCost(elapsed);
- if (batchMessageLogger.isDebugEnabled()) {
- batchMessageLogger.debug("batchMessage|eventMesh2mq|REQ|ASYNC|batchId={}|send2MQCost={}ms|msgNum={}|topics={}",
- batchId, elapsed, eventSize, topicBatchMessageMappings.keySet());
- }
+
+ LogUtils.debug(batchMessageLogger, "batchMessage|eventMesh2mq|REQ|ASYNC|batchId={}|send2MQCost={}ms|msgNum={}|topics={}",
+ batchId, elapsed, eventSize, topicBatchMessageMappings.keySet());
completeResponse(request, asyncContext, sendMessageBatchResponseHeader, EventMeshRetCode.SUCCESS, null,
SendMessageBatchResponseBody.class);
return;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
index a16578bbc..a390cfccb 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
@@ -31,6 +31,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchV2RequestHeader;
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageBatchV2ResponseHeader;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -197,9 +198,7 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
.withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP,
String.valueOf(System.currentTimeMillis()))
.build();
- if (batchMessageLogger.isDebugEnabled()) {
- batchMessageLogger.debug("msg2MQMsg suc, topic:{}, msg:{}", topic, event.getData());
- }
+ LogUtils.debug(batchMessageLogger, "msg2MQMsg suc, topic:{}, msg:{}", topic, event.getData());
} catch (Exception e) {
batchMessageLogger.error("msg2MQMsg err, topic:{}, msg:{}", topic, event.getData(), e);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
index 03e75266f..e146871ad 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/CreateTopicProcessor.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.common.EventMeshTrace;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -117,9 +118,7 @@ public class CreateTopicProcessor implements AsyncHttpProcessor {
final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper -> {
try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpEventWrapper);
- }
+ LogUtils.debug(httpLogger, "{}", httpEventWrapper);
eventMeshHTTPServer.sendResponse(ctx, httpEventWrapper.httpResponse());
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
System.currentTimeMillis() - requestWrapper.getReqTime());
@@ -151,6 +150,6 @@ public class CreateTopicProcessor implements AsyncHttpProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.CREATE_TOPIC.getRequestURI()};
+ return new String[] {RequestURI.CREATE_TOPIC.getRequestURI()};
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
index 0e2702ac2..6ad858b89 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/DeleteTopicProcessor.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.common.EventMeshTrace;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -140,9 +141,7 @@ public class DeleteTopicProcessor implements AsyncHttpProcessor {
final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper -> {
try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpEventWrapper);
- }
+ LogUtils.debug(httpLogger, "{}", httpEventWrapper);
eventMeshHTTPServer.sendResponse(ctx, httpEventWrapper.httpResponse());
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
System.currentTimeMillis() - requestWrapper.getReqTime());
@@ -174,6 +173,6 @@ public class DeleteTopicProcessor implements AsyncHttpProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.DELETE_TOPIC.getRequestURI()};
+ return new String[] {RequestURI.DELETE_TOPIC.getRequestURI()};
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
index b745e8f83..7fb0bde52 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
@@ -22,6 +22,7 @@ import org.apache.eventmesh.common.enums.ConnectionType;
import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.HTTPTrace;
import org.apache.eventmesh.runtime.boot.HTTPTrace.TraceOperation;
import org.apache.eventmesh.runtime.common.EventMeshTrace;
@@ -296,9 +297,7 @@ public class HandlerService {
private void postHandler(ConnectionType type) {
metrics.getSummaryMetrics().recordHTTPRequest();
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", request);
- }
+ LogUtils.debug(httpLogger, "{}", request);
if (Objects.isNull(response)) {
this.response = HttpResponseUtils.createSuccess();
}
@@ -312,9 +311,7 @@ public class HandlerService {
private void preHandler() {
metrics.getSummaryMetrics().recordHTTPReqResTimeCost(System.currentTimeMillis() - requestTime);
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", response);
- }
+ LogUtils.debug(httpLogger, "{}", response);
}
private void error() {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
index 50359a722..c6914d34f 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.http.header.client.HeartbeatRequestHeader;
import org.apache.eventmesh.common.protocol.http.header.client.HeartbeatResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
@@ -68,12 +69,10 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
HttpCommand responseEventMeshCommand;
final String localAddress = IPUtils.getLocalAddress();
HttpCommand request = asyncContext.getRequest();
- if (log.isInfoEnabled()) {
- log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
- RequestCode.get(Integer.valueOf(request.getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
- }
+ LogUtils.info(log, "cmd={}|{}|client2eventMesh|from={}|to={}",
+ RequestCode.get(Integer.valueOf(request.getRequestCode())),
+ EventMeshConstants.PROTOCOL_HTTP,
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
final HeartbeatRequestHeader heartbeatRequestHeader = (HeartbeatRequestHeader) request.getHeader();
final HeartbeatRequestBody heartbeatRequestBody = (HeartbeatRequestBody) request.getBody();
EventMeshHTTPConfiguration httpConfiguration = eventMeshHTTPServer.getEventMeshHttpConfiguration();
@@ -132,9 +131,7 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
} catch (Exception e) {
completeResponse(request, asyncContext, heartbeatResponseHeader,
EventMeshRetCode.EVENTMESH_ACL_ERR, e.getMessage(), HeartbeatResponseBody.class);
- if (log.isWarnEnabled()) {
- log.warn("CLIENT HAS NO PERMISSION,HeartBeatProcessor subscribe failed", e);
- }
+ LogUtils.warn(log, "CLIENT HAS NO PERMISSION,HeartBeatProcessor subscribe failed", e);
return;
}
}
@@ -167,9 +164,7 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
try {
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
- if (log.isDebugEnabled()) {
- log.debug("{}", httpCommand);
- }
+ LogUtils.debug(log, "{}", httpCommand);
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
summaryMetrics.recordHTTPReqResTimeCost(
System.currentTimeMillis() - request.getReqTime());
@@ -185,9 +180,7 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2),
HeartbeatResponseBody.class);
final long elapsedTime = System.currentTimeMillis() - startTime;
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|heartBeatMessageCost={}ms", elapsedTime, e);
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|heartBeatMessageCost={}ms", elapsedTime, e);
summaryMetrics.recordSendMsgFailed();
summaryMetrics.recordSendMsgCost(elapsedTime);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
index 1b15c1ac5..eb03c8dc8 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.common.EventMeshTrace;
@@ -67,10 +68,8 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
final HttpEventWrapper requestWrapper = handlerSpecific.getAsyncContext().getRequest();
String localAddress = IPUtils.getLocalAddress();
String remoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
- if (log.isInfoEnabled()) {
- log.info("uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
- }
+ LogUtils.info(log, "uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
// user request header
requestWrapper.getHeaderMap().put(ProtocolKey.ClientInstanceKey.IP.getKey(), remoteAddr);
@@ -120,9 +119,7 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
this.acl.doAclCheckInHttpReceive(remoteAddr, user, pass, subsystem, item.getTopic(),
requestWrapper.getRequestURI());
} catch (Exception e) {
- if (log.isWarnEnabled()) {
- log.warn("CLIENT HAS NO PERMISSION,SubscribeProcessor subscribe failed", e);
- }
+ LogUtils.warn(log, "CLIENT HAS NO PERMISSION,SubscribeProcessor subscribe failed", e);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_ACL_ERR, responseHeaderMap,
responseBodyMap, null);
@@ -135,18 +132,14 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
try {
if (!IPUtils.isValidDomainOrIp(url, eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIpv4BlackList(),
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshIpv6BlackList())) {
- if (log.isErrorEnabled()) {
- log.error("subscriber url {} is not valid", url);
- }
+ LogUtils.error(log, "subscriber url {} is not valid", url);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
responseBodyMap, null);
return;
}
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("subscriber url {} is not valid", url, e);
- }
+ LogUtils.error(log, "subscriber url {} is not valid", url, e);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
responseBodyMap, null);
@@ -156,9 +149,7 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
// obtain webhook delivery agreement for Abuse Protection
if (!WebhookUtil.obtainDeliveryAgreement(eventMeshHTTPServer.getHttpClientPool().getClient(),
url, eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshWebhookOrigin())) {
- if (log.isErrorEnabled()) {
- log.error("subscriber url {} is not allowed by the target system", url);
- }
+ LogUtils.error(log, "subscriber url {} is not allowed by the target system", url);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR, responseHeaderMap,
responseBodyMap, null);
return;
@@ -181,12 +172,10 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}",
- System.currentTimeMillis() - startTime,
- JsonUtils.toJSONString(subscriptionList),
- url, e);
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}",
+ System.currentTimeMillis() - startTime,
+ JsonUtils.toJSONString(subscriptionList),
+ url, e);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR, responseHeaderMap,
responseBodyMap, null);
@@ -200,7 +189,7 @@ public class LocalSubscribeEventProcessor extends AbstractEventProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.SUBSCRIBE_LOCAL.getRequestURI()};
+ return new String[] {RequestURI.SUBSCRIBE_LOCAL.getRequestURI()};
}
private ClientInfo getClientInfo(final HttpEventWrapper requestWrapper) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
index 204bb1873..aede5ebab 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.common.EventMeshTrace;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -76,10 +77,9 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
String localAddress = IPUtils.getLocalAddress();
String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- if (log.isInfoEnabled()) {
- log.info("uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
- }
+
+ LogUtils.info(log, "uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
// user request header
requestWrapper.getHeaderMap().put(ProtocolKey.ClientInstanceKey.IP.getKey(), remoteAddr);
@@ -136,9 +136,7 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
final Client client = clientIterator.next();
if (StringUtils.equals(client.getPid(), pid)
&& StringUtils.equals(client.getUrl(), unSubscribeUrl)) {
- if (log.isWarnEnabled()) {
- log.warn("client {} start unsubscribe", JsonUtils.toJSONString(client));
- }
+ LogUtils.warn(log, "client {} start unsubscribe", JsonUtils.toJSONString(client));
clientIterator.remove();
}
}
@@ -194,11 +192,9 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ "|topic={}|url={}", System.currentTimeMillis() - startTime,
- JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
- }
+ JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
responseBodyMap, null);
}
@@ -218,11 +214,9 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().keySet()
.removeIf(s -> StringUtils.equals(consumerGroup, s));
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
+ "|topic={}|url={}", System.currentTimeMillis() - startTime,
- JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
- }
+ JsonUtils.toJSONString(unSubTopicList), unSubscribeUrl, e);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR, responseHeaderMap,
responseBodyMap, null);
}
@@ -235,12 +229,10 @@ public class LocalUnSubscribeEventProcessor extends AbstractEventProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI()};
+ return new String[] {RequestURI.UNSUBSCRIBE_LOCAL.getRequestURI()};
}
- private void registerClient(final HttpEventWrapper requestWrapper,
- final String consumerGroup,
- final List<String> topicList, final String url) {
+ private void registerClient(final HttpEventWrapper requestWrapper, final String consumerGroup, final List<String> topicList, final String url) {
Objects.requireNonNull(requestWrapper, "requestWrapper can not be null");
Objects.requireNonNull(consumerGroup, "consumerGroup can not be null");
Objects.requireNonNull(topicList, "topicList can not be null");
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
index ac9455299..97629d489 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/QuerySubscriptionProcessor.java
@@ -22,6 +22,7 @@ import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.common.EventMeshTrace;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -78,9 +79,7 @@ public class QuerySubscriptionProcessor implements AsyncHttpProcessor {
final CompleteHandler<HttpEventWrapper> handler = httpEventWrapper -> {
try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpEventWrapper);
- }
+ LogUtils.debug(httpLogger, "{}", httpEventWrapper);
eventMeshHTTPServer.sendResponse(ctx, httpEventWrapper.httpResponse());
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
System.currentTimeMillis() - requestWrapper.getReqTime());
@@ -114,6 +113,6 @@ public class QuerySubscriptionProcessor implements AsyncHttpProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.SUBSCRIPTION_QUERY.getRequestURI()};
+ return new String[] {RequestURI.SUBSCRIPTION_QUERY.getRequestURI()};
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
index d3f9a2603..1c8ac67b3 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
@@ -31,6 +31,7 @@ import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.http.header.message.ReplyMessageRequestHeader;
import org.apache.eventmesh.common.protocol.http.header.message.ReplyMessageResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -175,9 +176,7 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
.withExtension(EventMeshConstants.REQ_C2EVENTMESH_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
.build();
- if (messageLogger.isDebugEnabled()) {
- messageLogger.debug("msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, replyTopic);
- }
+ LogUtils.debug(messageLogger, "msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, replyTopic);
} catch (Exception e) {
messageLogger.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, replyTopic, e);
@@ -192,9 +191,7 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
summaryMetrics.recordReplyMsg();
CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpCommand);
- }
+ LogUtils.debug(httpLogger, "{}", httpCommand);
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
summaryMetrics.recordHTTPReqResTimeCost(
System.currentTimeMillis() - request.getReqTime());
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java
index 69d7192de..6b0d178ec 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java
@@ -29,6 +29,7 @@ import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -81,10 +82,8 @@ public class SendAsyncEventProcessor implements AsyncHttpProcessor {
final String localAddress = IPUtils.getLocalAddress();
- if (log.isInfoEnabled()) {
- log.info("uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP, RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
- }
+ LogUtils.info(log, "uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
+ EventMeshConstants.PROTOCOL_HTTP, RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
// user request header
final Map<String, Object> requestHeaderMap = requestWrapper.getHeaderMap();
@@ -183,9 +182,7 @@ public class SendAsyncEventProcessor implements AsyncHttpProcessor {
} catch (Exception e) {
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_ACL_ERR, responseHeaderMap,
responseBodyMap, EventMeshUtil.getCloudEventExtensionMap(SpecVersion.V1.toString(), event));
- if (log.isWarnEnabled()) {
- log.warn("CLIENT HAS NO PERMISSION,SendAsyncMessageProcessor send failed", e);
- }
+ LogUtils.warn(log, "CLIENT HAS NO PERMISSION,SendAsyncMessageProcessor send failed", e);
return;
}
}
@@ -213,10 +210,8 @@ public class SendAsyncEventProcessor implements AsyncHttpProcessor {
final String content = new String(Objects.requireNonNull(event.getData()).toBytes(), StandardCharsets.UTF_8);
if (Objects.requireNonNull(content).length() > eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()) {
- if (log.isErrorEnabled()) {
- log.error("Event size exceeds the limit: {}",
- eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
- }
+ LogUtils.error(log, "Event size exceeds the limit: {}",
+ eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_SIZE_ERR, responseHeaderMap,
responseBodyMap, EventMeshUtil.getCloudEventExtensionMap(SpecVersion.V1.toString(), event));
return;
@@ -229,13 +224,9 @@ public class SendAsyncEventProcessor implements AsyncHttpProcessor {
.withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
.build();
- if (log.isDebugEnabled()) {
- log.debug("msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
- }
+ LogUtils.debug(log, "msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, topic, e);
- }
+ LogUtils.error(log, "msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, topic, e);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR, responseHeaderMap,
responseBodyMap, EventMeshUtil.getCloudEventExtensionMap(SpecVersion.V1.toString(), event));
return;
@@ -261,10 +252,8 @@ public class SendAsyncEventProcessor implements AsyncHttpProcessor {
responseBodyMap.put(EventMeshConstants.RET_CODE, EventMeshRetCode.SUCCESS.getRetCode());
responseBodyMap.put(EventMeshConstants.RET_MSG, EventMeshRetCode.SUCCESS.getErrMsg() + sendResult);
- if (log.isInfoEnabled()) {
- log.info("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- System.currentTimeMillis() - startTime, topic, bizNo, uniqueId);
- }
+ LogUtils.info(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ System.currentTimeMillis() - startTime, topic, bizNo, uniqueId);
handlerSpecific.getTraceOperation().endLatestTrace(sendMessageContext.getEvent());
handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
}
@@ -279,10 +268,8 @@ public class SendAsyncEventProcessor implements AsyncHttpProcessor {
EventMeshUtil.getCloudEventExtensionMap(SpecVersion.V1.toString(), sendMessageContext.getEvent()));
handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- System.currentTimeMillis() - startTime, topic, bizNo, uniqueId, context.getException());
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ System.currentTimeMillis() - startTime, topic, bizNo, uniqueId, context.getException());
}
});
} catch (Exception ex) {
@@ -290,15 +277,13 @@ public class SendAsyncEventProcessor implements AsyncHttpProcessor {
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR, responseHeaderMap, responseBodyMap, null);
final long endTime = System.currentTimeMillis();
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, topic, bizNo, uniqueId, ex);
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ endTime - startTime, topic, bizNo, uniqueId, ex);
}
}
@Override
public String[] paths() {
- return new String[]{RequestURI.PUBLISH.getRequestURI()};
+ return new String[] {RequestURI.PUBLISH.getRequestURI()};
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
index 6881034ac..9a71fdad3 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
@@ -31,6 +31,7 @@ import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageRequestHeader;
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -212,9 +213,7 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
.withExtension(EventMeshConstants.REQ_SEND_EVENTMESH_IP, eventMeshHttpConfiguration.getEventMeshServerIp())
.build();
- if (MESSAGE_LOGGER.isDebugEnabled()) {
- MESSAGE_LOGGER.debug("msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
- }
+ LogUtils.debug(MESSAGE_LOGGER, "msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
} catch (Exception e) {
MESSAGE_LOGGER.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, topic, e);
completeResponse(request, asyncContext, sendMessageResponseHeader,
@@ -233,9 +232,7 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
- if (HTTP_LOGGER.isDebugEnabled()) {
- HTTP_LOGGER.debug("{}", httpCommand);
- }
+ LogUtils.debug(HTTP_LOGGER, "{}", httpCommand);
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
summaryMetrics.recordHTTPReqResTimeCost(
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
index d5046f5e9..405f8ed38 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncRemoteEventProcessor.java
@@ -28,6 +28,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestURI;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -86,10 +87,8 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
final HttpEventWrapper requestWrapper = asyncContext.getRequest();
final String localAddress = IPUtils.getLocalAddress();
- if (log.isInfoEnabled()) {
- log.info("uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
- EventMeshConstants.PROTOCOL_HTTP, RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
- }
+ LogUtils.info(log, "uri={}|{}|client2eventMesh|from={}|to={}", requestWrapper.getRequestURI(),
+ EventMeshConstants.PROTOCOL_HTTP, RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
// user request header
final Map<String, Object> requestHeaderMap = requestWrapper.getHeaderMap();
@@ -244,10 +243,8 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
final String content = event.getData() == null ? "" : new String(event.getData().toBytes(), StandardCharsets.UTF_8);
if (content.length() > eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize()) {
- if (log.isErrorEnabled()) {
- log.error("Event size exceeds the limit: {}",
- eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
- }
+ LogUtils.error(log, "Event size exceeds the limit: {}",
+ eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshEventSize());
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_SIZE_ERR, responseHeaderMap,
responseBodyMap, EventMeshUtil.getCloudEventExtensionMap(SpecVersion.V1.toString(), event));
return;
@@ -260,13 +257,9 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
.withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
.build();
- if (log.isDebugEnabled()) {
- log.debug("msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
- }
+ LogUtils.debug(log, "msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, topic, e);
- }
+ LogUtils.error(log, "msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, topic, e);
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR, responseHeaderMap,
responseBodyMap, EventMeshUtil.getCloudEventExtensionMap(SpecVersion.V1.toString(), event));
return;
@@ -292,10 +285,8 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
responseBodyMap.put(EventMeshConstants.RET_CODE, EventMeshRetCode.SUCCESS.getRetCode());
responseBodyMap.put(EventMeshConstants.RET_MSG, EventMeshRetCode.SUCCESS.getErrMsg() + sendResult);
- if (log.isInfoEnabled()) {
- log.info("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- System.currentTimeMillis() - startTime, topic, bizNo, uniqueId);
- }
+ LogUtils.info(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ System.currentTimeMillis() - startTime, topic, bizNo, uniqueId);
handlerSpecific.getTraceOperation().endLatestTrace(sendMessageContext.getEvent());
handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
}
@@ -311,10 +302,8 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
handlerSpecific.sendResponse(responseHeaderMap, responseBodyMap);
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- System.currentTimeMillis() - startTime, topic, bizNo, uniqueId, context.getException());
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ System.currentTimeMillis() - startTime, topic, bizNo, uniqueId, context.getException());
}
});
} catch (Exception ex) {
@@ -322,10 +311,8 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
handlerSpecific.sendErrorResponse(EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR, responseHeaderMap,
responseBodyMap, null);
- if (log.isErrorEnabled()) {
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- System.currentTimeMillis() - startTime, topic, bizNo, uniqueId, ex);
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ System.currentTimeMillis() - startTime, topic, bizNo, uniqueId, ex);
}
}
@@ -337,7 +324,7 @@ public class SendAsyncRemoteEventProcessor implements AsyncHttpProcessor {
@Override
public String[] paths() {
- return new String[]{RequestURI.PUBLISH_BRIDGE.getRequestURI()};
+ return new String[] {RequestURI.PUBLISH_BRIDGE.getRequestURI()};
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
index e6902ebf9..ee1a1d834 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
@@ -30,6 +30,7 @@ import org.apache.eventmesh.common.protocol.http.header.message.SendMessageReque
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -77,11 +78,9 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
HttpCommand request = asyncContext.getRequest();
final String localAddress = IPUtils.getLocalAddress();
final String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- if (log.isInfoEnabled()) {
- log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
- RequestCode.get(Integer.valueOf(request.getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
- }
+ LogUtils.info(log, "cmd={}|{}|client2eventMesh|from={}|to={}",
+ RequestCode.get(Integer.valueOf(request.getRequestCode())),
+ EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
SendMessageRequestHeader sendMessageRequestHeader = (SendMessageRequestHeader) request.getHeader();
@@ -143,9 +142,7 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
} catch (Exception e) {
completeResponse(request, asyncContext, sendMessageResponseHeader,
EventMeshRetCode.EVENTMESH_ACL_ERR, e.getMessage(), SendMessageResponseBody.class);
- if (log.isWarnEnabled()) {
- log.warn("CLIENT HAS NO PERMISSION,SendSyncMessageProcessor send failed", e);
- }
+ LogUtils.warn(log, "CLIENT HAS NO PERMISSION,SendSyncMessageProcessor send failed", e);
return;
}
}
@@ -163,9 +160,7 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
final String content = new String(Objects.requireNonNull(event.getData()).toBytes(), Constants.DEFAULT_CHARSET);
int eventMeshEventSize = eventMeshHttpConfiguration.getEventMeshEventSize();
if (content.length() > eventMeshEventSize) {
- if (log.isErrorEnabled()) {
- log.error("Event size exceeds the limit: {}", eventMeshEventSize);
- }
+ LogUtils.error(log, "Event size exceeds the limit: {}", eventMeshEventSize);
completeResponse(request, asyncContext, sendMessageResponseHeader,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
"Event size exceeds the limit: " + eventMeshEventSize,
@@ -190,14 +185,10 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
.withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
.build();
- if (log.isDebugEnabled()) {
- log.debug("msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
- }
+ LogUtils.debug(log, "msg2MQMsg suc, bizSeqNo={}, topic={}", bizNo, topic);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, topic, e);
- }
+ LogUtils.error(log, "msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, topic, e);
completeResponse(request, asyncContext, sendMessageResponseHeader,
EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR, null, SendMessageResponseBody.class);
return;
@@ -211,9 +202,7 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
- if (log.isDebugEnabled()) {
- log.debug("{}", httpCommand);
- }
+ LogUtils.debug(log, "{}", httpCommand);
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHTTPReqResTimeCost(
System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
@@ -228,11 +217,9 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
@Override
public void onSuccess(final CloudEvent event) {
- if (log.isInfoEnabled()) {
- log.info("message|mq2eventMesh|RSP|SYNC|rrCost={}ms|topic={}"
+ LogUtils.info(log, "message|mq2eventMesh|RSP|SYNC|rrCost={}ms|topic={}"
+ "|bizSeqNo={}|uniqueId={}", System.currentTimeMillis() - startTime,
- topic, bizNo, uniqueId);
- }
+ topic, bizNo, uniqueId);
try {
final CloudEvent newEvent = CloudEventBuilder.from(event)
@@ -263,9 +250,7 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
+ EventMeshUtil.stackTrace(ex, 2)));
asyncContext.onComplete(err, handler);
- if (log.isWarnEnabled()) {
- log.warn("message|mq2eventMesh|RSP", ex);
- }
+ LogUtils.warn(log, "message|mq2eventMesh|RSP", ex);
}
}
@@ -280,13 +265,10 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
asyncContext.onComplete(err, handler);
eventMeshHTTPServer.getHttpRetryer().newTimeout(sendMessageContext, 10, TimeUnit.SECONDS);
- if (log.isErrorEnabled()) {
- log.error(
- "message|mq2eventMesh|RSP|SYNC|rrCost={}ms|topic={}"
- + "|bizSeqNo={}|uniqueId={}",
- System.currentTimeMillis() - startTime,
- topic, bizNo, uniqueId, e);
- }
+ LogUtils.error(log, "message|mq2eventMesh|RSP|SYNC|rrCost={}ms|topic={}"
+ + "|bizSeqNo={}|uniqueId={}",
+ System.currentTimeMillis() - startTime,
+ topic, bizNo, uniqueId, e);
}
}, Integer.parseInt(ttl));
} catch (Exception ex) {
@@ -299,11 +281,8 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
summaryMetrics.recordSendMsgFailed();
summaryMetrics.recordSendMsgCost(endTime - startTime);
- if (log.isErrorEnabled()) {
- log.error(
- "message|eventMesh2mq|REQ|SYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, topic, bizNo, uniqueId, ex);
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|SYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ endTime - startTime, topic, bizNo, uniqueId, ex);
}
return;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
index a6c7ff622..f2bfeb1c6 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
@@ -27,6 +27,7 @@ import org.apache.eventmesh.common.protocol.http.header.client.SubscribeRequestH
import org.apache.eventmesh.common.protocol.http.header.client.SubscribeResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
@@ -70,10 +71,8 @@ public class SubscribeProcessor implements HttpRequestProcessor {
final Integer requestCode = Integer.valueOf(request.getRequestCode());
final String localAddress = IPUtils.getLocalAddress();
final String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- if (log.isInfoEnabled()) {
- log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
- RequestCode.get(requestCode), EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
- }
+ LogUtils.info(log, "cmd={}|{}|client2eventMesh|from={}|to={}",
+ RequestCode.get(requestCode), EventMeshConstants.PROTOCOL_HTTP, remoteAddr, localAddress);
final SubscribeRequestHeader subscribeRequestHeader = (SubscribeRequestHeader) request.getHeader();
final SubscribeRequestBody subscribeRequestBody = (SubscribeRequestBody) request.getBody();
@@ -116,9 +115,7 @@ public class SubscribeProcessor implements HttpRequestProcessor {
} catch (Exception e) {
completeResponse(request, asyncContext, subscribeResponseHeader,
EventMeshRetCode.EVENTMESH_ACL_ERR, e.getMessage(), SubscribeResponseBody.class);
- if (log.isWarnEnabled()) {
- log.warn("CLIENT HAS NO PERMISSION,SubscribeProcessor subscribe failed", e);
- }
+ LogUtils.warn(log, "CLIENT HAS NO PERMISSION,SubscribeProcessor subscribe failed", e);
return;
}
}
@@ -139,9 +136,7 @@ public class SubscribeProcessor implements HttpRequestProcessor {
return;
}
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("subscriber url:{} is invalid.", url, e);
- }
+ LogUtils.error(log, "subscriber url:{} is invalid.", url, e);
completeResponse(request, asyncContext, subscribeResponseHeader,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR,
EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg() + " invalid URL: " + url,
@@ -175,9 +170,7 @@ public class SubscribeProcessor implements HttpRequestProcessor {
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
- if (log.isDebugEnabled()) {
- log.debug("{}", httpCommand);
- }
+ LogUtils.debug(log, "{}", httpCommand);
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
@@ -194,14 +187,11 @@ public class SubscribeProcessor implements HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2),
SubscribeResponseBody.class);
final long endTime = System.currentTimeMillis();
- if (log.isErrorEnabled()) {
- log.error(
- "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}"
- + "|bizSeqNo={}|uniqueId={}",
- endTime - startTime,
- JsonUtils.toJSONString(subscribeRequestBody.getTopics()),
- subscribeRequestBody.getUrl(), e);
- }
+ LogUtils.error(log, "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}"
+ + "|bizSeqNo={}|uniqueId={}",
+ endTime - startTime,
+ JsonUtils.toJSONString(subscribeRequestBody.getTopics()),
+ subscribeRequestBody.getUrl(), e);
summaryMetrics.recordSendMsgFailed();
summaryMetrics.recordSendMsgCost(endTime - startTime);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
index 4e9277b0a..18ba3354e 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.common.protocol.http.header.client.UnSubscribeReques
import org.apache.eventmesh.common.protocol.http.header.client.UnSubscribeResponseHeader;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
@@ -72,12 +73,11 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
HttpCommand responseEventMeshCommand;
final HttpCommand request = asyncContext.getRequest();
final String localAddress = IPUtils.getLocalAddress();
- if (log.isInfoEnabled()) {
- log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
- RequestCode.get(Integer.valueOf(request.getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
- }
+
+ LogUtils.info(log, "cmd={}|{}|client2eventMesh|from={}|to={}",
+ RequestCode.get(Integer.valueOf(request.getRequestCode())),
+ EventMeshConstants.PROTOCOL_HTTP,
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()), localAddress);
final UnSubscribeRequestHeader unSubscribeRequestHeader = (UnSubscribeRequestHeader) request.getHeader();
final UnSubscribeRequestBody unSubscribeRequestBody = (UnSubscribeRequestBody) request.getBody();
@@ -113,9 +113,7 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
HttpSummaryMetrics summaryMetrics = eventMeshHTTPServer.getMetrics().getSummaryMetrics();
final CompleteHandler<HttpCommand> handler = httpCommand -> {
try {
- if (log.isDebugEnabled()) {
- log.debug("{}", httpCommand);
- }
+ LogUtils.debug(log, "{}", httpCommand);
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
} catch (Exception ex) {
@@ -140,9 +138,7 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
final Client client = clientIterator.next();
if (StringUtils.equals(client.getPid(), pid)
&& StringUtils.equals(client.getUrl(), unSubscribeUrl)) {
- if (log.isWarnEnabled()) {
- log.warn("client {} start unsubscribe", JsonUtils.toJSONString(client));
- }
+ LogUtils.warn(log, "client {} start unsubscribe", JsonUtils.toJSONString(client));
clientIterator.remove();
}
}
@@ -198,10 +194,8 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2),
UnSubscribeResponseBody.class);
final long endTime = System.currentTimeMillis();
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}",
- endTime - startTime,
- JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()),
- unSubscribeRequestBody.getUrl(), e);
+ log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}", endTime - startTime,
+ JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()), unSubscribeRequestBody.getUrl(), e);
summaryMetrics.recordSendMsgFailed();
summaryMetrics.recordSendMsgCost(endTime - startTime);
}
@@ -225,10 +219,8 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
EventMeshRetCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2),
UnSubscribeResponseBody.class);
final long endTime = System.currentTimeMillis();
- log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms"
- + "|topic={}|url={}", endTime - startTime,
- JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()),
- unSubscribeRequestBody.getUrl(), e);
+ log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|url={}", endTime - startTime,
+ JsonUtils.toJSONString(unSubscribeRequestBody.getTopics()), unSubscribeRequestBody.getUrl(), e);
summaryMetrics.recordSendMsgFailed();
summaryMetrics.recordSendMsgCost(endTime - startTime);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
index 213bfeef2..747db0c9e 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
@@ -29,6 +29,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
@@ -78,6 +79,7 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
public static final Logger LOGGER = LoggerFactory.getLogger("AsyncHTTPPushRequest");
public String currPushUrl;
+
private final Map<String, Set<AbstractHTTPPushRequest>> waitingRequests;
public AsyncHTTPPushRequest(HandleMsgContext handleMsgContext,
@@ -195,10 +197,8 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
addToWaitingMap(this);
- if (CMD_LOGGER.isInfoEnabled()) {
- CMD_LOGGER.info("cmd={}|eventMesh2client|from={}|to={}", requestCode,
- localAddress, currPushUrl);
- }
+ LogUtils.info(CMD_LOGGER, "cmd={}|eventMesh2client|from={}|to={}", requestCode,
+ localAddress, currPushUrl);
try {
eventMeshHTTPServer.getHttpClientPool().getClient().execute(builder, response -> {
@@ -240,13 +240,10 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
}
} else {
eventMeshHTTPServer.getMetrics().getSummaryMetrics().recordHttpPushMsgFailed();
- if (MESSAGE_LOGGER.isInfoEnabled()) {
- MESSAGE_LOGGER.info(
- "message|eventMesh2client|exception|url={}|topic={}|bizSeqNo={}"
- + "|uniqueId={}|cost={}",
- currPushUrl, handleMsgContext.getTopic(),
- handleMsgContext.getBizSeqNo(), handleMsgContext.getUniqueId(), cost);
- }
+ LogUtils.info(MESSAGE_LOGGER, "message|eventMesh2client|exception|url={}|topic={}|bizSeqNo={}"
+ + "|uniqueId={}|cost={}",
+ currPushUrl, handleMsgContext.getTopic(),
+ handleMsgContext.getBizSeqNo(), handleMsgContext.getUniqueId(), cost);
if (isComplete()) {
handleMsgContext.finish();
@@ -335,16 +332,12 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
return ClientRetCode.FAIL;
} catch (NumberFormatException e) {
- if (MESSAGE_LOGGER.isWarnEnabled()) {
- MESSAGE_LOGGER.warn("url:{}, bizSeqno:{}, uniqueId:{}, httpResponse:{}", currPushUrl,
- handleMsgContext.getBizSeqNo(), handleMsgContext.getUniqueId(), content);
- }
+ LogUtils.warn(MESSAGE_LOGGER, "url:{}, bizSeqno:{}, uniqueId:{}, httpResponse:{}", currPushUrl,
+ handleMsgContext.getBizSeqNo(), handleMsgContext.getUniqueId(), content);
return ClientRetCode.FAIL;
} catch (Exception e) {
- if (MESSAGE_LOGGER.isWarnEnabled()) {
- MESSAGE_LOGGER.warn("url:{}, bizSeqno:{}, uniqueId:{}, httpResponse:{}", currPushUrl,
- handleMsgContext.getBizSeqNo(), handleMsgContext.getUniqueId(), content);
- }
+ LogUtils.warn(MESSAGE_LOGGER, "url:{}, bizSeqno:{}, uniqueId:{}, httpResponse:{}", currPushUrl,
+ handleMsgContext.getBizSeqNo(), handleMsgContext.getUniqueId(), content);
return ClientRetCode.FAIL;
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
index 61b264806..72b1927fa 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
@@ -28,6 +28,7 @@ import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -217,15 +218,11 @@ public class ClientGroupWrapper {
}
Session s = topic2sessionInGroupMapping.get(topic).putIfAbsent(session.getSessionId(), session);
if (s == null) {
- if (log.isInfoEnabled()) {
- log.info("Cache session success, group:{} topic:{} client:{} sessionId:{}", group,
- topic, session.getClient(), session.getSessionId());
- }
+ LogUtils.info(log, "Cache session success, group:{} topic:{} client:{} sessionId:{}", group,
+ topic, session.getClient(), session.getSessionId());
} else {
- if (log.isWarnEnabled()) {
- log.warn("Session already exists in topic2sessionInGroupMapping. group:{} topic:{} client:{} sessionId:{}", group, topic,
- session.getClient(), session.getSessionId());
- }
+ LogUtils.warn(log, "Session already exists in topic2sessionInGroupMapping. group:{} topic:{} client:{} sessionId:{}", group, topic,
+ session.getClient(), session.getSessionId());
}
subscriptions.putIfAbsent(topic, subscriptionItem);
@@ -247,9 +244,7 @@ public class ClientGroupWrapper {
return false;
}
String topic = subscriptionItem.getTopic();
- if (session == null
- || !StringUtils.equalsIgnoreCase(group,
- EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
+ if (session == null || !StringUtils.equalsIgnoreCase(group, EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
log.error("removeSubscription param error,topic:{},session:{}", topic, session);
return false;
}
@@ -330,10 +325,7 @@ public class ClientGroupWrapper {
}
public boolean addGroupConsumerSession(Session session) {
- if (session == null
- || !StringUtils.equalsIgnoreCase(group,
- EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
-
+ if (session == null || !StringUtils.equalsIgnoreCase(group, EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
log.error("addGroupConsumerSession param error,session:{}", session);
return false;
}
@@ -362,10 +354,7 @@ public class ClientGroupWrapper {
}
public boolean addGroupProducerSession(Session session) {
- if (session == null
- || !StringUtils.equalsIgnoreCase(group,
- EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
-
+ if (session == null || !StringUtils.equalsIgnoreCase(group, EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
log.error("addGroupProducerSession param error,session:{}", session);
return false;
}
@@ -392,10 +381,7 @@ public class ClientGroupWrapper {
}
public boolean removeGroupConsumerSession(Session session) {
- if (session == null
- || !StringUtils.equalsIgnoreCase(group,
- EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
-
+ if (session == null || !StringUtils.equalsIgnoreCase(group, EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
log.error("removeGroupConsumerSession param error,session:{}", session);
return false;
}
@@ -424,9 +410,7 @@ public class ClientGroupWrapper {
}
public boolean removeGroupProducerSession(Session session) {
- if (session == null
- || !StringUtils.equalsIgnoreCase(group,
- EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
+ if (session == null || !StringUtils.equalsIgnoreCase(group, EventMeshUtil.buildClientGroup(session.getClient().getGroup()))) {
log.error("removeGroupProducerSession param error,session:{}", session);
return false;
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
index 86279daef..f6ac3d73c 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
@@ -22,6 +22,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -375,9 +376,7 @@ public class ClientSessionGroupMapping {
long interval = System.currentTimeMillis() - tmp.getLastHeartbeatTime();
if (interval > eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshTcpSessionExpiredInMills()) {
try {
- if (log.isWarnEnabled()) {
- log.warn("clean expired session,client:{}", tmp.getClient());
- }
+ LogUtils.warn(log, "clean expired session,client:{}", tmp.getClient());
closeSession(tmp.getContext());
} catch (Exception e) {
log.error("say goodbye to session error! {}", tmp, e);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/dispatch/FreePriorityDispatchStrategy.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/dispatch/FreePriorityDispatchStrategy.java
index 68561347d..f725d8158 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/dispatch/FreePriorityDispatchStrategy.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/dispatch/FreePriorityDispatchStrategy.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.runtime.core.protocol.tcp.client.group.dispatch;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
import org.apache.commons.collections4.CollectionUtils;
@@ -49,10 +50,8 @@ public class FreePriorityDispatchStrategy implements DownstreamDispatchStrategy
if (session.isIsolated()) {
isolatedSessions.add(session);
- if (log.isInfoEnabled()) {
- log.info("session is not available because session is isolated,isolateTime:{},client:{}",
- session.getIsolateTime(), session.getClient());
- }
+ LogUtils.info(log, "session is not available because session is isolated,isolateTime:{},client:{}",
+ session.getIsolateTime(), session.getClient());
continue;
}
@@ -61,14 +60,10 @@ public class FreePriorityDispatchStrategy implements DownstreamDispatchStrategy
if (CollectionUtils.isEmpty(filtered)) {
if (CollectionUtils.isEmpty(isolatedSessions)) {
- if (log.isWarnEnabled()) {
- log.warn("all sessions can't downstream msg");
- }
+ LogUtils.warn(log, "all sessions can't downstream msg");
return null;
} else {
- if (log.isWarnEnabled()) {
- log.warn("all sessions are isolated,group:{},topic:{}", group, topic);
- }
+ LogUtils.warn(log, "all sessions are isolated,group:{},topic:{}", group, topic);
filtered.addAll(isolatedSessions);
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java
index 61364e47a..a216561c5 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java
@@ -91,7 +91,7 @@ public class HelloProcessor implements TcpProcessor {
pkg.getHeader().getSeq()));
Utils.writeAndFlush(res, startTime, taskExecuteTime, session.getContext(), session);
} catch (Throwable e) {
- MESSAGE_LOGGER.error("HelloTask failed|address={},errMsg={}", ctx.channel().remoteAddress(), e);
+ MESSAGE_LOGGER.error("HelloTask failed|address={}", ctx.channel().remoteAddress(), e);
res.setHeader(new Header(HELLO_RESPONSE, OPStatus.FAIL.getCode(), Arrays.toString(e.getStackTrace()), pkg
.getHeader().getSeq()));
ctx.writeAndFlush(res).addListener(
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java
index 273ef17c2..83b0c2652 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.protocol.tcp.Header;
import org.apache.eventmesh.common.protocol.tcp.OPStatus;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.Subscription;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
@@ -42,8 +43,8 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class SubscribeProcessor implements TcpProcessor {
- private EventMeshTCPServer eventMeshTCPServer;
private final Acl acl;
+ private EventMeshTCPServer eventMeshTCPServer;
public SubscribeProcessor(EventMeshTCPServer eventMeshTCPServer) {
this.eventMeshTCPServer = eventMeshTCPServer;
@@ -86,9 +87,7 @@ public class SubscribeProcessor implements TcpProcessor {
synchronized (session) {
session.subscribe(subscriptionItems);
- if (log.isInfoEnabled()) {
- log.info("SubscribeTask succeed|user={}|topics={}", session.getClient(), subscriptionItems);
- }
+ LogUtils.info(log, "SubscribeTask succeed|user={}|topics={}", session.getClient(), subscriptionItems);
}
eventMeshTCPServer.getClientSessionGroupMapping().updateMetaData();
msg.setHeader(new Header(Command.SUBSCRIBE_RESPONSE, OPStatus.SUCCESS.getCode(), OPStatus.SUCCESS.getDesc(), pkg.getHeader().getSeq()));
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
index bbf673d09..159a05dff 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend;
import org.apache.eventmesh.api.meta.dto.EventMeshDataInfo;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.apache.eventmesh.runtime.util.ValueComparator;
@@ -50,9 +51,7 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
List<EventMeshDataInfo> eventMeshDataInfoList;
if (StringUtils.isAnyBlank(group, purpose)) {
- if (log.isWarnEnabled()) {
- log.warn("EventMeshRecommend failed,params illegal,group:{},purpose:{}", group, purpose);
- }
+ LogUtils.warn(log, "EventMeshRecommend failed,params illegal,group:{},purpose:{}", group, purpose);
return null;
}
@@ -60,18 +59,14 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
try {
eventMeshDataInfoList = eventMeshTCPServer.getMetaStorage().findEventMeshInfoByCluster(cluster);
} catch (Exception e) {
- if (log.isWarnEnabled()) {
- log.warn("EventMeshRecommend failed, findEventMeshInfoByCluster failed, cluster:{}, group:{}, purpose:{}, errMsg:{}",
- cluster, group, purpose, e);
- }
+ LogUtils.warn(log, "EventMeshRecommend failed, findEventMeshInfoByCluster failed, cluster:{}, group:{}, purpose:{}, errMsg:{}",
+ cluster, group, purpose, e);
return null;
}
if (CollectionUtils.isEmpty(eventMeshDataInfoList)) {
- if (log.isWarnEnabled()) {
- log.warn("EventMeshRecommend failed,not find eventMesh instances from registry,cluster:{},group:{},purpose:{}",
- cluster, group, purpose);
- }
+ LogUtils.warn(log, "EventMeshRecommend failed,not find eventMesh instances from registry,cluster:{},group:{},purpose:{}",
+ cluster, group, purpose);
return null;
}
@@ -85,9 +80,7 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
? localEventMeshMap.put(eventMeshDataInfo.getEventMeshName(), eventMeshDataInfo.getEndpoint())
: remoteEventMeshMap.put(eventMeshDataInfo.getEventMeshName(), eventMeshDataInfo.getEndpoint());
} else {
- if (log.isErrorEnabled()) {
- log.error("EventMeshName may be illegal,idc is null,eventMeshName:{}", eventMeshDataInfo.getEventMeshName());
- }
+ LogUtils.error(log, "EventMeshName may be illegal,idc is null,eventMeshName:{}", eventMeshDataInfo.getEventMeshName());
}
}
@@ -116,10 +109,8 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
return new ArrayList<String>();
}
- if (log.isInfoEnabled()) {
- log.info("eventMeshMap:{},clientDistributionMap:{},group:{},recommendNum:{},currEventMeshName:{}",
- eventMeshMap, clientDistributedMap, group, recommendProxyNum, eventMeshName);
- }
+ LogUtils.info(log, "eventMeshMap:{},clientDistributionMap:{},group:{},recommendNum:{},currEventMeshName:{}",
+ eventMeshMap, clientDistributedMap, group, recommendProxyNum, eventMeshName);
// find eventmesh with least client
final List<Map.Entry<String, Integer>> clientDistributedList = new ArrayList<>();
@@ -127,9 +118,7 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
clientDistributedMap.entrySet().forEach(clientDistributedList::add);
Collections.sort(clientDistributedList, vc);
- if (log.isInfoEnabled()) {
- log.info("clientDistributedLists after sort:{}", clientDistributedList);
- }
+ LogUtils.info(log, "clientDistributedLists after sort:{}", clientDistributedList);
final List<String> recommendProxyList = new ArrayList<>(recommendProxyNum);
while (recommendProxyList.size() < recommendProxyNum) {
@@ -139,15 +128,11 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
clientDistributedMap.put(minProxyItem.getKey(), minProxyItem.getValue() + 1);
clientDistributedMap.put(eventMeshName, currProxyNum - 1);
Collections.sort(clientDistributedList, vc);
- if (log.isInfoEnabled()) {
- log.info("clientDistributedList after sort:{}", clientDistributedList);
- }
+ LogUtils.info(log, "clientDistributedList after sort:{}", clientDistributedList);
}
- if (log.isInfoEnabled()) {
- log.info("choose proxys with min instance num, group:{}, recommendProxyNum:{}, recommendProxyList:{}",
- group, recommendProxyNum, recommendProxyList);
- }
+ LogUtils.info(log, "choose proxys with min instance num, group:{}, recommendProxyNum:{}, recommendProxyList:{}",
+ group, recommendProxyNum, recommendProxyList);
return recommendProxyList;
}
@@ -155,20 +140,16 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
final Map<String, String> eventMeshMap, final boolean caculateLocal) {
Objects.requireNonNull(eventMeshMap, "eventMeshMap can not be null");
- if (log.isInfoEnabled()) {
- log.info("eventMeshMap:{},cluster:{},group:{},purpose:{},caculateLocal:{}", eventMeshMap, cluster,
- group, purpose, caculateLocal);
- }
+ LogUtils.info(log, "eventMeshMap:{},cluster:{},group:{},purpose:{},caculateLocal:{}", eventMeshMap, cluster,
+ group, purpose, caculateLocal);
Map<String, Map<String, Integer>> eventMeshClientDistributionDataMap = null;
try {
eventMeshClientDistributionDataMap = eventMeshTCPServer.getMetaStorage().findEventMeshClientDistributionData(
cluster, group, purpose);
} catch (Exception e) {
- if (log.isWarnEnabled()) {
- log.warn("EventMeshRecommend failed,findEventMeshClientDistributionData failed,"
- + "cluster:{},group:{},purpose:{}, errMsg:{}", cluster, group, purpose, e);
- }
+ LogUtils.warn(log, "EventMeshRecommend failed,findEventMeshClientDistributionData failed,"
+ + "cluster:{},group:{},purpose:{}, errMsg:{}", cluster, group, purpose, e);
}
String recommendProxyAddr;
@@ -180,10 +161,8 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
Collections.shuffle(tmpProxyAddrList);
recommendProxyAddr = tmpProxyAddrList.get(0);
- if (log.isInfoEnabled()) {
- log.info("No distribute data in registry,cluster:{}, group:{},purpose:{}, recommendProxyAddr:{}",
- cluster, group, purpose, recommendProxyAddr);
- }
+ LogUtils.info(log, "No distribute data in registry,cluster:{}, group:{},purpose:{}, recommendProxyAddr:{}",
+ cluster, group, purpose, recommendProxyAddr);
return recommendProxyAddr;
}
@@ -199,19 +178,15 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
remoteClientDistributionMap.put(entry.getKey(), entry.getValue().get(purpose));
}
} else {
- if (log.isErrorEnabled()) {
- log.error("eventMeshName may be illegal,idc is null,eventMeshName:{}", entry.getKey());
- }
+ LogUtils.error(log, "eventMeshName may be illegal,idc is null,eventMeshName:{}", entry.getKey());
}
});
recommendProxyAddr = recommendProxy(eventMeshMap, (caculateLocal == true) ? localClientDistributionMap
: remoteClientDistributionMap, group);
- if (log.isInfoEnabled()) {
- log.info("eventMeshMap:{},group:{},purpose:{},caculateLocal:{},recommendProxyAddr:{}", eventMeshMap,
- group, purpose, caculateLocal, recommendProxyAddr);
- }
+ LogUtils.info(log, "eventMeshMap:{},group:{},purpose:{},caculateLocal:{},recommendProxyAddr:{}", eventMeshMap,
+ group, purpose, caculateLocal, recommendProxyAddr);
return recommendProxyAddr;
}
@@ -222,14 +197,10 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
Objects.requireNonNull(eventMeshMap, "eventMeshMap can not be null");
Objects.requireNonNull(clientDistributionMap, "clientDistributionMap can not be null");
- if (log.isInfoEnabled()) {
- log.info("eventMeshMap:{},clientDistributionMap:{},group:{}", eventMeshMap, clientDistributionMap, group);
- }
+ LogUtils.info(log, "eventMeshMap:{},clientDistributionMap:{},group:{}", eventMeshMap, clientDistributionMap, group);
if (!eventMeshMap.keySet().containsAll(clientDistributionMap.keySet())) {
- if (log.isWarnEnabled()) {
- log.warn("exist proxy not register but exist in distributionMap");
- }
+ LogUtils.warn(log, "exist proxy not register but exist in distributionMap");
return null;
}
@@ -237,17 +208,13 @@ public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
// select the eventmesh with least instances
if (MapUtils.isEmpty(clientDistributionMap)) {
- if (log.isErrorEnabled()) {
- log.error("no legal distribute data,check eventMeshMap and distributeData, group:{}", group);
- }
+ LogUtils.error(log, "no legal distribute data,check eventMeshMap and distributeData, group:{}", group);
return null;
} else {
final List<Map.Entry<String, Integer>> list = new ArrayList<>();
clientDistributionMap.entrySet().forEach(list::add);
Collections.sort(list, new ValueComparator());
- if (log.isInfoEnabled()) {
- log.info("clientDistributionMap after sort:{}", list);
- }
+ LogUtils.info(log, "clientDistributionMap after sort:{}", list);
return eventMeshMap.get(list.get(0).getKey());
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
index 8bd740898..4d26c0d2f 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.metrics.http;
import org.apache.eventmesh.common.EventMeshThreadFactory;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.metrics.api.MetricsRegistry;
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.metrics.api.model.RetrySummaryMetrics;
@@ -58,17 +59,13 @@ public class HTTPMetricsServer {
private void init() {
metricsRegistries.forEach(MetricsRegistry::start);
- if (log.isInfoEnabled()) {
- log.info("HTTPMetricsServer initialized.");
- }
+ LogUtils.info(log, "HTTPMetricsServer initialized.");
}
public void start() {
metricsRegistries.forEach(metricsRegistry -> {
metricsRegistry.register(summaryMetrics);
- if (log.isInfoEnabled()) {
- log.info("Register httpMetrics to {}", metricsRegistry.getClass().getName());
- }
+ LogUtils.info(log, "Register httpMetrics to {}", metricsRegistry.getClass().getName());
});
metricsSchedule.scheduleAtFixedRate(() -> {
@@ -90,17 +87,13 @@ public class HTTPMetricsServer {
}
}, 1000, 30 * 1000, TimeUnit.MILLISECONDS);
- if (log.isInfoEnabled()) {
- log.info("HTTPMetricsServer started.");
- }
+ LogUtils.info(log, "HTTPMetricsServer started.");
}
public void shutdown() {
metricsSchedule.shutdown();
metricsRegistries.forEach(MetricsRegistry::showdown);
- if (log.isInfoEnabled()) {
- log.info("HTTPMetricsServer shutdown.");
- }
+ LogUtils.info(log, "HTTPMetricsServer shutdown.");
}
private static ScheduledExecutorService metricsSchedule = Executors.newScheduledThreadPool(2,
@@ -110,73 +103,59 @@ public class HTTPMetricsServer {
private void logPrintServerMetrics(final HttpSummaryMetrics summaryMetrics,
final EventMeshHTTPServer eventMeshHTTPServer) {
- if (log.isInfoEnabled()) {
- log.info("===========================================SERVER METRICS==================================================");
-
- log.info("maxHTTPTPS: {}, avgHTTPTPS: {}, maxHTTPCOST: {}, avgHTTPCOST: {}, avgHTTPBodyDecodeCost: {}, httpDiscard: {}",
- summaryMetrics.maxHTTPTPS(),
- summaryMetrics.avgHTTPTPS(),
- summaryMetrics.maxHTTPCost(),
- summaryMetrics.avgHTTPCost(),
- summaryMetrics.avgHTTPBodyDecodeCost(),
- summaryMetrics.getHttpDiscard());
- }
+ LogUtils.info(log, "===========================================SERVER METRICS==================================================");
+ LogUtils.info(log, "maxHTTPTPS: {}, avgHTTPTPS: {}, maxHTTPCOST: {}, avgHTTPCOST: {}, avgHTTPBodyDecodeCost: {}, httpDiscard: {}",
+ summaryMetrics.maxHTTPTPS(),
+ summaryMetrics.avgHTTPTPS(),
+ summaryMetrics.maxHTTPCost(),
+ summaryMetrics.avgHTTPCost(),
+ summaryMetrics.avgHTTPBodyDecodeCost(),
+ summaryMetrics.getHttpDiscard());
summaryMetrics.httpStatInfoClear();
- if (log.isInfoEnabled()) {
- log.info("maxBatchSendMsgTPS: {}, avgBatchSendMsgTPS: {}, sum: {}. sumFail: {}, sumFailRate: {}, discard : {}",
- summaryMetrics.maxSendBatchMsgTPS(),
- summaryMetrics.avgSendBatchMsgTPS(),
- summaryMetrics.getSendBatchMsgNumSum(),
- summaryMetrics.getSendBatchMsgFailNumSum(),
- summaryMetrics.getSendBatchMsgFailRate(),
- summaryMetrics.getSendBatchMsgDiscardNumSum());
- }
+ LogUtils.info(log, "maxBatchSendMsgTPS: {}, avgBatchSendMsgTPS: {}, sum: {}. sumFail: {}, sumFailRate: {}, discard : {}",
+ summaryMetrics.maxSendBatchMsgTPS(),
+ summaryMetrics.avgSendBatchMsgTPS(),
+ summaryMetrics.getSendBatchMsgNumSum(),
+ summaryMetrics.getSendBatchMsgFailNumSum(),
+ summaryMetrics.getSendBatchMsgFailRate(),
+ summaryMetrics.getSendBatchMsgDiscardNumSum());
summaryMetrics.cleanSendBatchStat();
- if (log.isInfoEnabled()) {
- log.info("maxSendMsgTPS: {}, avgSendMsgTPS: {}, sum: {}, sumFail: {}, sumFailRate: {}, replyMsg: {}, replyFail: {}",
- summaryMetrics.maxSendMsgTPS(),
- summaryMetrics.avgSendMsgTPS(),
- summaryMetrics.getSendMsgNumSum(),
- summaryMetrics.getSendMsgFailNumSum(),
- summaryMetrics.getSendMsgFailRate(),
- summaryMetrics.getReplyMsgNumSum(),
- summaryMetrics.getReplyMsgFailNumSum());
- }
+ LogUtils.info(log, "maxSendMsgTPS: {}, avgSendMsgTPS: {}, sum: {}, sumFail: {}, sumFailRate: {}, replyMsg: {}, replyFail: {}",
+ summaryMetrics.maxSendMsgTPS(),
+ summaryMetrics.avgSendMsgTPS(),
+ summaryMetrics.getSendMsgNumSum(),
+ summaryMetrics.getSendMsgFailNumSum(),
+ summaryMetrics.getSendMsgFailRate(),
+ summaryMetrics.getReplyMsgNumSum(),
+ summaryMetrics.getReplyMsgFailNumSum());
summaryMetrics.cleanSendMsgStat();
- if (log.isInfoEnabled()) {
- log.info(
- "maxPushMsgTPS: {}, avgPushMsgTPS: {}, sum: {}, sumFail: {}, sumFailRate: {}, maxClientLatency: {}, avgClientLatency: {}",
- summaryMetrics.maxPushMsgTPS(),
- summaryMetrics.avgPushMsgTPS(),
- summaryMetrics.getHttpPushMsgNumSum(),
- summaryMetrics.getHttpPushFailNumSum(),
- summaryMetrics.getHttpPushMsgFailRate(),
- summaryMetrics.maxHTTPPushLatency(),
- summaryMetrics.avgHTTPPushLatency());
- }
+ LogUtils.info(log, "maxPushMsgTPS: {}, avgPushMsgTPS: {}, sum: {}, sumFail: {}, sumFailRate: {}, maxClientLatency: {}, avgClientLatency: {}",
+ summaryMetrics.maxPushMsgTPS(),
+ summaryMetrics.avgPushMsgTPS(),
+ summaryMetrics.getHttpPushMsgNumSum(),
+ summaryMetrics.getHttpPushFailNumSum(),
+ summaryMetrics.getHttpPushMsgFailRate(),
+ summaryMetrics.maxHTTPPushLatency(),
+ summaryMetrics.avgHTTPPushLatency());
summaryMetrics.cleanHttpPushMsgStat();
- if (log.isInfoEnabled()) {
- log.info("batchMsgQ: {}, sendMsgQ: {}, pushMsgQ: {}, httpRetryQ: {}",
- eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor().getQueue().size(),
- eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor().getQueue().size(),
- eventMeshHTTPServer.getHttpThreadPoolGroup().getPushMsgExecutor().getQueue().size(),
- eventMeshHTTPServer.getHttpRetryer().getPendingTimeouts());
- }
-
- if (log.isInfoEnabled()) {
- log.info("batchAvgSend2MQCost: {}, avgSend2MQCost: {}, avgReply2MQCost: {}",
- summaryMetrics.avgBatchSendMsgCost(),
- summaryMetrics.avgSendMsgCost(),
- summaryMetrics.avgReplyMsgCost());
- }
+ LogUtils.info(log, "batchMsgQ: {}, sendMsgQ: {}, pushMsgQ: {}, httpRetryQ: {}",
+ eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor().getQueue().size(),
+ eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor().getQueue().size(),
+ eventMeshHTTPServer.getHttpThreadPoolGroup().getPushMsgExecutor().getQueue().size(),
+ eventMeshHTTPServer.getHttpRetryer().getPendingTimeouts());
+
+ LogUtils.info(log, "batchAvgSend2MQCost: {}, avgSend2MQCost: {}, avgReply2MQCost: {}",
+ summaryMetrics.avgBatchSendMsgCost(),
+ summaryMetrics.avgSendMsgCost(),
+ summaryMetrics.avgReplyMsgCost());
summaryMetrics.send2MQStatInfoClear();
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java
index 6404cb576..324464cbe 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.runtime.util;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -170,9 +171,7 @@ public class EventMeshUtil {
public static String getLocalAddr() {
// priority of networkInterface when generating client ip
final String priority = System.getProperty("networkInterface.priority", "bond1<eth1<eth0");
- if (log.isDebugEnabled()) {
- log.debug("networkInterface.priority: {}", priority);
- }
+ LogUtils.debug(log, "networkInterface.priority: {}", priority);
final List<String> preferList = new ArrayList<>();
preferList.addAll(Arrays.asList(priority.split("<")));
@@ -196,15 +195,11 @@ public class EventMeshUtil {
final ArrayList<String> ipv6Result = new ArrayList<>();
if (preferNetworkInterface != null) {
- if (log.isDebugEnabled()) {
- log.debug("use preferNetworkInterface:{}", preferNetworkInterface);
- }
+ LogUtils.debug(log, "use preferNetworkInterface:{}", preferNetworkInterface);
final Enumeration<InetAddress> en = preferNetworkInterface.getInetAddresses();
getIpResult(ipv4Result, ipv6Result, en);
} else {
- if (log.isDebugEnabled()) {
- log.debug("no preferNetworkInterface");
- }
+ LogUtils.debug(log, "no preferNetworkInterface");
final Enumeration<NetworkInterface> enumeration = NetworkInterface.getNetworkInterfaces();
while (enumeration.hasMoreElements()) {
final NetworkInterface networkInterface = enumeration.nextElement();
@@ -277,11 +272,9 @@ public class EventMeshUtil {
}
public static void printState(final ThreadPoolExecutor scheduledExecutorService) {
- if (log.isInfoEnabled()) {
- log.info("{} [{} {} {} {}]", ((EventMeshThreadFactory) scheduledExecutorService.getThreadFactory())
+ LogUtils.info(log, "{} [{} {} {} {}]", ((EventMeshThreadFactory) scheduledExecutorService.getThreadFactory())
.getThreadNamePrefix(), scheduledExecutorService.getQueue().size(), scheduledExecutorService.getPoolSize(),
- scheduledExecutorService.getActiveCount(), scheduledExecutorService.getCompletedTaskCount());
- }
+ scheduledExecutorService.getActiveCount(), scheduledExecutorService.getCompletedTaskCount());
}
/**
@@ -307,8 +300,7 @@ public class EventMeshUtil {
}
- public static Map<String, Object> getCloudEventExtensionMap(final String protocolVersion,
- final CloudEvent cloudEvent) {
+ public static Map<String, Object> getCloudEventExtensionMap(final String protocolVersion, final CloudEvent cloudEvent) {
final EventMeshCloudEventWriter eventMeshCloudEventWriter = new EventMeshCloudEventWriter();
if (StringUtils.equals(SpecVersion.V1.toString(), protocolVersion)
&& cloudEvent instanceof CloudEventV1) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/WebhookUtil.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/WebhookUtil.java
index 8308fdad5..38a418e4b 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/WebhookUtil.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/WebhookUtil.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.util;
import org.apache.eventmesh.api.auth.AuthService;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.spi.EventMeshExtensionFactory;
import org.apache.commons.lang3.StringUtils;
@@ -52,9 +53,7 @@ public class WebhookUtil {
final String targetUrl,
final String requestOrigin) {
- if (log.isInfoEnabled()) {
- log.info("obtain webhook delivery agreement for url: {}", targetUrl);
- }
+ LogUtils.info(log, "obtain webhook delivery agreement for url: {}", targetUrl);
final HttpOptions builder = new HttpOptions(targetUrl);
builder.addHeader(REQUEST_ORIGIN_HEADER, requestOrigin);
@@ -68,10 +67,8 @@ public class WebhookUtil {
return StringUtils.isEmpty(allowedOrigin)
|| "*".equals(allowedOrigin) || allowedOrigin.equalsIgnoreCase(requestOrigin);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("HTTP Options Method is not supported at the Delivery Target: {}, "
- + "unable to obtain the webhook delivery agreement.", targetUrl);
- }
+ LogUtils.error(log, "HTTP Options Method is not supported at the Delivery Target: {}, "
+ + "unable to obtain the webhook delivery agreement.", targetUrl);
}
return true;
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/RequestContext.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/RequestContext.java
index 0a9179580..6155b2942 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/RequestContext.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/RequestContext.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.client.common;
import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.utils.LogUtils;
import java.util.concurrent.CountDownLatch;
@@ -76,9 +77,7 @@ public class RequestContext {
public static RequestContext context(Object key, Package request, CountDownLatch latch) throws Exception {
RequestContext c = new RequestContext(key, request, latch);
- if (log.isInfoEnabled()) {
- log.info("_RequestContext|create|key=" + key);
- }
+ LogUtils.info(log, "_RequestContext|create|key=", key);
return c;
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/PubClientImpl.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/PubClientImpl.java
index 20faf76af..0ef3699cf 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/PubClientImpl.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/PubClientImpl.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.OPStatus;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.api.PubClient;
import org.apache.eventmesh.runtime.client.common.ClientConstants;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
@@ -60,9 +61,7 @@ public class PubClientImpl extends TCPClient implements PubClient {
public void init() throws Exception {
open(new Handler());
hello();
- if (log.isInfoEnabled()) {
- log.info("PubClientImpl|{}|started!", clientNo);
- }
+ LogUtils.info(log, "PubClientImpl|{}|started!", clientNo);
}
public void reconnect() throws Exception {
@@ -86,10 +85,8 @@ public class PubClientImpl extends TCPClient implements PubClient {
PubClientImpl.this.reconnect();
}
Package msg = MessageUtils.heartBeat();
- if (log.isDebugEnabled()) {
- log.debug("PubClientImpl|{}|send heartbeat|Command={}|msg={}",
- clientNo, msg.getHeader().getCommand(), msg);
- }
+ LogUtils.debug(log, "PubClientImpl|{}|send heartbeat|Command={}|msg={}",
+ clientNo, msg.getHeader().getCommand(), msg);
PubClientImpl.this.dispatcher(msg, ClientConstants.DEFAULT_TIMEOUT_IN_MILLISECONDS);
} catch (Exception ignored) {
// ignore
@@ -117,9 +114,7 @@ public class PubClientImpl extends TCPClient implements PubClient {
*/
@Override
public Package rr(Package msg, long timeout) throws Exception {
- if (log.isInfoEnabled()) {
- log.info("PubClientImpl|{}|rr|send|Command={}|msg={}", clientNo, Command.REQUEST_TO_SERVER, msg);
- }
+ LogUtils.info(log, "PubClientImpl|{}|rr|send|Command={}|msg={}", clientNo, Command.REQUEST_TO_SERVER, msg);
return dispatcher(msg, timeout);
}
@@ -164,9 +159,7 @@ public class PubClientImpl extends TCPClient implements PubClient {
* Send an event message, the return value is ACCESS and ACK is given
*/
public Package publish(Package msg, long timeout) throws Exception {
- if (log.isInfoEnabled()) {
- log.info("PubClientImpl|{}|publish|send|command={}|msg={}", clientNo, msg.getHeader().getCommand(), msg);
- }
+ LogUtils.info(log, "PubClientImpl|{}|publish|send|command={}|msg={}", clientNo, msg.getHeader().getCommand(), msg);
return dispatcher(msg, timeout);
}
@@ -174,9 +167,7 @@ public class PubClientImpl extends TCPClient implements PubClient {
* send broadcast message
*/
public Package broadcast(Package msg, long timeout) throws Exception {
- if (log.isInfoEnabled()) {
- log.info("PubClientImpl|{}|broadcast|send|type={}|msg={}", clientNo, msg.getHeader().getCommand(), msg);
- }
+ LogUtils.info(log, "PubClientImpl|{}|broadcast|send|type={}|msg={}", clientNo, msg.getHeader().getCommand(), msg);
return dispatcher(msg, timeout);
}
@@ -190,9 +181,7 @@ public class PubClientImpl extends TCPClient implements PubClient {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Package msg) throws Exception {
- if (log.isInfoEnabled()) {
- log.info("PubClientImpl|{}|receive|type={}|msg={}", clientNo, msg.getHeader().getCommand(), msg);
- }
+ LogUtils.info(log, "PubClientImpl|{}|receive|type={}|msg={}", clientNo, msg.getHeader().getCommand(), msg);
Command cmd = msg.getHeader().getCommand();
if (callback != null) {
callback.handle(msg, ctx);
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/SubClientImpl.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/SubClientImpl.java
index bd37aed1f..c9e389b3e 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/SubClientImpl.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/impl/SubClientImpl.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.OPStatus;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.api.SubClient;
import org.apache.eventmesh.runtime.client.common.ClientConstants;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
@@ -69,9 +70,7 @@ public class SubClientImpl extends TCPClient implements SubClient {
public void init() throws Exception {
open(new Handler());
hello();
- if (log.isInfoEnabled()) {
- log.info("SubClientImpl|{}|started!", clientNo);
- }
+ LogUtils.info(log, "SubClientImpl|{}|started!", clientNo);
}
public void reconnect() throws Exception {
@@ -102,10 +101,8 @@ public class SubClientImpl extends TCPClient implements SubClient {
SubClientImpl.this.reconnect();
}
Package msg = MessageUtils.heartBeat();
- if (log.isDebugEnabled()) {
- log.debug("SubClientImpl|{}|send heartbeat|Command={}|msg={}", clientNo,
- msg.getHeader().getCommand(), msg);
- }
+ LogUtils.debug(log, "SubClientImpl|{}|send heartbeat|Command={}|msg={}", clientNo,
+ msg.getHeader().getCommand(), msg);
SubClientImpl.this.dispatcher(msg, ClientConstants.DEFAULT_TIMEOUT_IN_MILLISECONDS);
} catch (Exception e) {
// ignore
@@ -141,7 +138,7 @@ public class SubClientImpl extends TCPClient implements SubClient {
* Package msg = MessageUtils.traceLog();
* this.dispatcher(msg, ClientConstants.DEFAULT_TIMEOUT_IN_MILLISECONDS);
* }
- **/
+ */
/**
*
@@ -203,10 +200,8 @@ public class SubClientImpl extends TCPClient implements SubClient {
@SuppressWarnings("Duplicates")
@Override
protected void channelRead0(ChannelHandlerContext ctx, Package msg) throws Exception {
- if (log.isInfoEnabled()) {
- log.info(SubClientImpl.class.getSimpleName() + "|receive|command={}|msg={}",
- msg.getHeader().getCommand(), msg);
- }
+ LogUtils.info(log, SubClientImpl.class.getSimpleName() + "|receive|command={}|msg={}",
+ msg.getHeader().getCommand(), msg);
Command cmd = msg.getHeader().getCommand();
if (callback != null) {
callback.handle(msg, ctx);
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java
index e46fe3e23..03801ef15 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.runtime.demo;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.client.common.ClientConstants;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
@@ -34,9 +35,7 @@ public class AsyncPubClient {
pubClient.init();
pubClient.heartbeat();
pubClient.registerBusiHandler((msg, ctx) -> {
- if (log.isInfoEnabled()) {
- log.info("server good by request: {}", msg);
- }
+ LogUtils.info(log, "server good by request: {}", msg);
});
for (int i = 0; i < 1; i++) {
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncSubClient.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncSubClient.java
index 4c5435e3a..e6a91b4af 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncSubClient.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncSubClient.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.runtime.demo;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.common.ClientConstants;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
import org.apache.eventmesh.runtime.client.impl.SubClientImpl;
@@ -38,9 +39,7 @@ public class AsyncSubClient {
client.registerBusiHandler((msg, ctx) -> {
if (msg.getBody() instanceof EventMeshMessage) {
String body = ((EventMeshMessage) msg.getBody()).getBody();
- if (log.isInfoEnabled()) {
- log.info("receive message : {}", body);
- }
+ LogUtils.info(log, "receive message : {}", body);
}
});
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastSubClient.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastSubClient.java
index e064bfe5a..a94e5a97c 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastSubClient.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastSubClient.java
@@ -22,6 +22,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.common.ClientConstants;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
import org.apache.eventmesh.runtime.client.hook.ReceiveMsgHook;
@@ -46,9 +47,7 @@ public class BroadCastSubClient {
if (msg.getHeader().getCommand() == Command.BROADCAST_MESSAGE_TO_CLIENT) {
if (msg.getBody() instanceof EventMeshMessage) {
String body = ((EventMeshMessage) msg.getBody()).getBody();
- if (log.isInfoEnabled()) {
- log.info("receive message -------------------------------" + body);
- }
+ LogUtils.info(log, "receive message -------------------------------{}", body);
}
}
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java
index e7cba1d22..6ceaeb3d5 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
import org.apache.eventmesh.runtime.client.common.UserAgentUtils;
import org.apache.eventmesh.runtime.client.impl.SubClientImpl;
@@ -38,9 +39,7 @@ public class CCSubClient {
subClient.listen();
subClient.justSubscribe("TEST-TOPIC-TCP-SYNC", SubscriptionMode.CLUSTERING, SubscriptionType.SYNC);
subClient.registerBusiHandler((msg, ctx) -> {
- if (log.isInfoEnabled()) {
- log.info("Received message: {}", msg);
- }
+ LogUtils.info(log, "Received message: {}", msg);
if (msg.getHeader().getCommand() == Command.REQUEST_TO_CLIENT) {
Package rrResponse = MessageUtils.rrResponse(msg);
ctx.writeAndFlush(rrResponse);
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CClientDemo.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CClientDemo.java
index 9500df952..4bd0b7ad9 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CClientDemo.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CClientDemo.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.runtime.demo;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.Command;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
import org.apache.eventmesh.runtime.client.impl.EventMeshClientImpl;
@@ -43,9 +44,7 @@ public class CClientDemo {
client.listen();
client.registerSubBusiHandler((msg, ctx) -> {
if (msg.getHeader().getCmd() == Command.ASYNC_MESSAGE_TO_CLIENT || msg.getHeader().getCmd() == Command.BROADCAST_MESSAGE_TO_CLIENT) {
- if (log.isInfoEnabled()) {
- log.info("receive message: {}", msg);
- }
+ LogUtils.info(log, "receive message: {}", msg);
}
});
for (int i = 0; i < 10000; i++) {
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncPubClient.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncPubClient.java
index 80e050a70..cb783028c 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncPubClient.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncPubClient.java
@@ -19,6 +19,7 @@ package org.apache.eventmesh.runtime.demo;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
import org.apache.eventmesh.runtime.client.common.UserAgentUtils;
import org.apache.eventmesh.runtime.client.impl.PubClientImpl;
@@ -38,10 +39,8 @@ public class SyncPubClient {
Package rr = pubClient.rr(MessageUtils.rrMesssage("TEST-TOPIC-TCP-SYNC", i), 3000);
if (rr.getBody() instanceof EventMeshMessage) {
String body = ((EventMeshMessage) rr.getBody()).getBody();
- if (log.isInfoEnabled()) {
- log.info("rrMessage: " + body + " "
- + "rr-reply-------------------------------------------------" + rr);
- }
+ LogUtils.info(log, "rrMessage: ", body, " ",
+ "rr-reply-------------------------------------------------{}", rr);
}
}
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncSubClient.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncSubClient.java
index e4a573426..c7a5474ed 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncSubClient.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/SyncSubClient.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.runtime.demo;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.common.protocol.tcp.Command;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.client.common.ClientConstants;
import org.apache.eventmesh.runtime.client.common.MessageUtils;
import org.apache.eventmesh.runtime.client.impl.SubClientImpl;
@@ -37,9 +38,7 @@ public class SyncSubClient {
client.justSubscribe(ClientConstants.SYNC_TOPIC, SubscriptionMode.CLUSTERING, SubscriptionType.SYNC);
client.registerBusiHandler((msg, ctx) -> {
if (msg.getHeader().getCommand() == Command.REQUEST_TO_CLIENT) {
- if ((log.isInfoEnabled())) {
- log.info("receive message:{}", msg);
- }
+ LogUtils.info(log, "receive message:{}", msg);
}
});
}
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/catalog/EventMeshCatalogClient.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/catalog/EventMeshCatalogClient.java
index ab58f9905..23224b4cf 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/catalog/EventMeshCatalogClient.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/catalog/EventMeshCatalogClient.java
@@ -28,6 +28,7 @@ import org.apache.eventmesh.common.protocol.catalog.protos.Operation;
import org.apache.eventmesh.common.protocol.catalog.protos.QueryOperationsRequest;
import org.apache.eventmesh.common.protocol.catalog.protos.QueryOperationsResponse;
import org.apache.eventmesh.common.utils.AssertUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.collections4.CollectionUtils;
@@ -66,9 +67,7 @@ public class EventMeshCatalogClient {
final QueryOperationsRequest request = QueryOperationsRequest.newBuilder()
.setServiceName(clientConfig.getAppServerName()).build();
final QueryOperationsResponse response = catalogClient.queryOperations(request);
- if (log.isInfoEnabled()) {
- log.info("received response: {}", response);
- }
+ LogUtils.info(log, "received response: {}", response);
final List<Operation> operations = response.getOperationsList();
if (CollectionUtils.isEmpty(operations)) {
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
index f85557d0f..83a33317e 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
@@ -41,6 +41,7 @@ import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.grpc.common.Response;
import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.collections4.MapUtils;
@@ -92,9 +93,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
}
public Response subscribe(final List<SubscriptionItem> subscriptionItems, final String url) {
- if (log.isInfoEnabled()) {
- log.info("Create subscription: {} , url: {}", subscriptionItems, url);
- }
+ LogUtils.info(log, "Create subscription: {} , url: {}", subscriptionItems, url);
addSubscription(subscriptionItems, url);
@@ -102,9 +101,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
url, subscriptionItems);
try {
CloudEvent response = consumerClient.subscribe(subscription);
- if (log.isInfoEnabled()) {
- log.info("Received response:{}", response);
- }
+ LogUtils.info(log, "Received response:{}", response);
return Response.builder()
.respCode(EventMeshCloudEventUtils.getResponseCode(response))
.respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
@@ -117,9 +114,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
}
public void subscribe(final List<SubscriptionItem> subscriptionItems) {
- if (log.isInfoEnabled()) {
- log.info("Create streaming subscription: {}", subscriptionItems);
- }
+ LogUtils.info(log, "Create streaming subscription: {}", subscriptionItems);
if (listener == null) {
log.error("Error in subscriber, no Event Listener is registered.");
@@ -151,9 +146,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
}
public Response unsubscribe(final List<SubscriptionItem> subscriptionItems, final String url) {
- if (log.isInfoEnabled()) {
- log.info("Removing subscription: {}, url:{}", subscriptionItems, url);
- }
+ LogUtils.info(log, "Removing subscription: {}, url:{}", subscriptionItems, url);
removeSubscription(subscriptionItems);
@@ -161,9 +154,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
subscriptionItems);
try {
final CloudEvent response = consumerClient.unsubscribe(cloudEvent);
- if (log.isInfoEnabled()) {
- log.info("Received response:{}", response);
- }
+ LogUtils.info(log, "Received response:{}", response);
return Response.builder()
.respCode(EventMeshCloudEventUtils.getResponseCode(response))
.respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
@@ -177,9 +168,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
public Response unsubscribe(final List<SubscriptionItem> subscriptionItems) {
Objects.requireNonNull(subscriptionItems, "subscriptionItems can not be null");
- if (log.isInfoEnabled()) {
- log.info("Removing subscription stream: {}", subscriptionItems);
- }
+ LogUtils.info(log, "Removing subscription stream: {}", subscriptionItems);
removeSubscription(subscriptionItems);
@@ -193,9 +182,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
.respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
.respTime(EventMeshCloudEventUtils.getResponseTime(response))
.build();
- if (log.isInfoEnabled()) {
- log.info("Received response:{}", parsedResponse);
- }
+ LogUtils.info(log, "Received response:{}", parsedResponse);
// there is no stream subscriptions, stop the subscription stream handler
synchronized (this) {
@@ -242,9 +229,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
.respMsg(EventMeshCloudEventUtils.getResponseMessage(cloudEventResp))
.respTime(EventMeshCloudEventUtils.getResponseTime(cloudEventResp))
.build();
- if (log.isDebugEnabled()) {
- log.debug("Grpc Consumer Heartbeat cloudEvent: {}", response);
- }
+ LogUtils.debug(log, "Grpc Consumer Heartbeat cloudEvent: {}", response);
if (StatusCode.CLIENT_RESUBSCRIBE.getRetCode().equals(response.getRespCode())) {
resubscribe();
}
@@ -253,9 +238,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
}
}, 10_000, EventMeshCommon.HEARTBEAT, TimeUnit.MILLISECONDS);
- if (log.isInfoEnabled()) {
- log.info("Grpc Consumer Heartbeat started.");
- }
+ LogUtils.info(log, "Grpc Consumer Heartbeat started.");
}
private void resubscribe() {
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
index 2c9f15792..282979315 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
@@ -27,6 +27,7 @@ import org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.grpc.common.SubscriptionReply;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import java.io.Serializable;
import java.util.HashMap;
@@ -53,7 +54,7 @@ public class SubStreamHandler<T> extends Thread implements Serializable {
private final ReceiveMsgHook<T> listener;
public SubStreamHandler(final ConsumerServiceStub consumerAsyncClient, final EventMeshGrpcClientConfig clientConfig,
- final ReceiveMsgHook<T> listener) {
+ final ReceiveMsgHook<T> listener) {
this.consumerAsyncClient = consumerAsyncClient;
this.clientConfig = clientConfig;
this.listener = listener;
@@ -75,14 +76,11 @@ public class SubStreamHandler<T> extends Thread implements Serializable {
public void onNext(final CloudEvent message) {
T msg = EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(message, listener.getProtocolType());
if (msg instanceof Set) {
- if (log.isInfoEnabled()) {
- log.info("Received message from Server:{}", message);
- }
+ LogUtils.info(log, "Received message from Server:{}", message);
} else {
- if (log.isInfoEnabled()) {
- log.info("Received message from Server.|seq={}|uniqueId={}|", EventMeshCloudEventUtils.getSeqNum(message),
- EventMeshCloudEventUtils.getUniqueId(message));
- }
+ LogUtils.info(log, "Received message from Server.|seq={}|uniqueId={}|",
+ EventMeshCloudEventUtils.getSeqNum(message),
+ EventMeshCloudEventUtils.getUniqueId(message));
CloudEvent streamReply = null;
try {
Optional<T> reply = listener.handle(msg);
@@ -90,17 +88,14 @@ public class SubStreamHandler<T> extends Thread implements Serializable {
streamReply = buildReplyMessage(message, reply.get());
}
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("Error in handling reply message.|seq={}|uniqueId={}|",
- EventMeshCloudEventUtils.getSeqNum(message), EventMeshCloudEventUtils.getUniqueId(message), e);
- }
+ LogUtils.error(log, "Error in handling reply message.|seq={}|uniqueId={}|",
+ EventMeshCloudEventUtils.getSeqNum(message),
+ EventMeshCloudEventUtils.getUniqueId(message), e);
}
if (streamReply != null) {
- if (log.isInfoEnabled()) {
- log.info("Sending reply message to Server.|seq={}|uniqueId={}|",
- EventMeshCloudEventUtils.getSeqNum(streamReply),
- EventMeshCloudEventUtils.getUniqueId(streamReply));
- }
+ LogUtils.info(log, "Sending reply message to Server.|seq={}|uniqueId={}|",
+ EventMeshCloudEventUtils.getSeqNum(streamReply),
+ EventMeshCloudEventUtils.getUniqueId(streamReply));
senderOnNext(streamReply);
}
}
@@ -108,17 +103,13 @@ public class SubStreamHandler<T> extends Thread implements Serializable {
@Override
public void onError(final Throwable t) {
- if (log.isErrorEnabled()) {
- log.error("Received Server side error", t);
- }
+ LogUtils.error(log, "Received Server side error", t);
close();
}
@Override
public void onCompleted() {
- if (log.isInfoEnabled()) {
- log.info("Finished receiving messages from server.");
- }
+ LogUtils.info(log, "Finished receiving messages from server.");
close();
}
};
@@ -164,9 +155,7 @@ public class SubStreamHandler<T> extends Thread implements Serializable {
latch.countDown();
- if (log.isInfoEnabled()) {
- log.info("SubStreamHandler closed.");
- }
+ LogUtils.info(log, "SubStreamHandler closed.");
}
private void senderOnNext(final CloudEvent subscription) {
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/CloudEventProducer.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/CloudEventProducer.java
index 4df9e64b2..e77f5bf14 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/CloudEventProducer.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/CloudEventProducer.java
@@ -25,6 +25,7 @@ import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEventBatch;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.PublisherServiceGrpc.PublisherServiceBlockingStub;
import org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
import org.apache.eventmesh.common.protocol.grpc.common.Response;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.collections4.CollectionUtils;
@@ -49,9 +50,7 @@ public class CloudEventProducer implements GrpcProducer<io.cloudevents.CloudEven
@Override
public Response publish(final List<io.cloudevents.CloudEvent> events) {
- if (log.isInfoEnabled()) {
- log.info("BatchPublish message, batch size={}", events.size());
- }
+ LogUtils.info(log, "BatchPublish message, batch size={}", events.size());
if (CollectionUtils.isEmpty(events)) {
return null;
@@ -65,23 +64,18 @@ public class CloudEventProducer implements GrpcProducer<io.cloudevents.CloudEven
.respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
.respTime(EventMeshCloudEventUtils.getResponseTime(response))
.build();
- if (log.isInfoEnabled()) {
- log.info("Received response:{}", parsedResponse);
- }
+
+ LogUtils.info(log, "Received response:{}", parsedResponse);
return parsedResponse;
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("Error in BatchPublish message {}", events, e);
- }
+ LogUtils.error(log, "Error in BatchPublish message {}", events, e);
}
return null;
}
@Override
public Response publish(final io.cloudevents.CloudEvent cloudEvent) {
- if (log.isInfoEnabled()) {
- log.info("Publish message: {}", cloudEvent.toString());
- }
+ LogUtils.info(log, "Publish message: {}", cloudEvent.toString());
CloudEvent enhancedMessage = EventMeshCloudEventBuilder.buildEventMeshCloudEvent(cloudEvent, clientConfig, PROTOCOL_TYPE);
try {
final CloudEvent response = publisherClient.publish(enhancedMessage);
@@ -90,34 +84,24 @@ public class CloudEventProducer implements GrpcProducer<io.cloudevents.CloudEven
.respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
.respTime(EventMeshCloudEventUtils.getResponseTime(response))
.build();
- if (log.isInfoEnabled()) {
- log.info("Received response:{} ", parsedResponse);
- }
+ LogUtils.info(log, "Received response:{} ", parsedResponse);
return parsedResponse;
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("Error in publishing message {}", cloudEvent, e);
- }
+ LogUtils.error(log, "Error in publishing message {}", cloudEvent, e);
}
return null;
}
@Override
public io.cloudevents.CloudEvent requestReply(final io.cloudevents.CloudEvent cloudEvent, final long timeout) {
- if (log.isInfoEnabled()) {
- log.info("RequestReply message {}", cloudEvent);
- }
+ LogUtils.info(log, "RequestReply message {}", cloudEvent);
final CloudEvent enhancedMessage = EventMeshCloudEventBuilder.buildEventMeshCloudEvent(cloudEvent, clientConfig, PROTOCOL_TYPE);
try {
final CloudEvent reply = publisherClient.requestReply(enhancedMessage);
- if (log.isInfoEnabled()) {
- log.info("Received reply message:{}", reply);
- }
+ LogUtils.info(log, "Received reply message:{}", reply);
return EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(reply, PROTOCOL_TYPE);
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("Error in RequestReply message {}", cloudEvent, e);
- }
+ LogUtils.error(log, "Error in RequestReply message {}", cloudEvent, e);
}
return null;
}
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshGrpcProducer.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshGrpcProducer.java
index b72b85be5..03d215595 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshGrpcProducer.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshGrpcProducer.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.common.EventMeshMessage;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.PublisherServiceGrpc;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.PublisherServiceGrpc.PublisherServiceBlockingStub;
import org.apache.eventmesh.common.protocol.grpc.common.Response;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.collections4.CollectionUtils;
@@ -60,9 +61,7 @@ public class EventMeshGrpcProducer implements AutoCloseable {
}
public <T> Response publish(T message) {
- if (log.isInfoEnabled()) {
- log.info("Publish message " + message.toString());
- }
+ LogUtils.info(log, "Publish message ", message.toString());
if (message instanceof CloudEvent) {
return cloudEventProducer.publish((CloudEvent) message);
} else if (message instanceof EventMeshMessage) {
@@ -74,9 +73,7 @@ public class EventMeshGrpcProducer implements AutoCloseable {
@SuppressWarnings("unchecked")
public <T> Response publish(List<T> messageList) {
- if (log.isInfoEnabled()) {
- log.info("BatchPublish message :{}", messageList);
- }
+ LogUtils.info(log, "BatchPublish message :{}", messageList);
if (CollectionUtils.isEmpty(messageList)) {
return null;
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshMessageProducer.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshMessageProducer.java
index 6ae647277..f7f2c7712 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshMessageProducer.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/producer/EventMeshMessageProducer.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEventBatch;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.PublisherServiceGrpc.PublisherServiceBlockingStub;
import org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
import org.apache.eventmesh.common.protocol.grpc.common.Response;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.collections4.CollectionUtils;
@@ -55,9 +56,7 @@ public class EventMeshMessageProducer implements GrpcProducer<EventMeshMessage>
return null;
}
- if (log.isDebugEnabled()) {
- log.info("Publish message: {}", message);
- }
+ LogUtils.debug(log, "Publish message: {}", message);
CloudEvent cloudEvent = EventMeshCloudEventBuilder.buildEventMeshCloudEvent(message, clientConfig, PROTOCOL_TYPE);
try {
CloudEvent response = publisherClient.publish(cloudEvent);
@@ -66,9 +65,7 @@ public class EventMeshMessageProducer implements GrpcProducer<EventMeshMessage>
.respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
.respTime(EventMeshCloudEventUtils.getResponseTime(response))
.build();
- if (log.isInfoEnabled()) {
- log.info("Received response:{}", parsedResponse);
- }
+ LogUtils.info(log, "Received response:{}", parsedResponse);
return parsedResponse;
} catch (Exception e) {
log.error("Error in publishing message {}", message, e);
@@ -90,30 +87,22 @@ public class EventMeshMessageProducer implements GrpcProducer<EventMeshMessage>
.respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
.respTime(EventMeshCloudEventUtils.getResponseTime(response))
.build();
- if (log.isInfoEnabled()) {
- log.info("Received response:{}", parsedResponse);
- }
+ LogUtils.info(log, "Received response:{}", parsedResponse);
return parsedResponse;
} catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("Error in BatchPublish message {}", messages, e);
- }
+ LogUtils.error(log, "Error in BatchPublish message {}", messages, e);
}
return null;
}
@Override
public EventMeshMessage requestReply(EventMeshMessage message, long timeout) {
- if (log.isInfoEnabled()) {
- log.info("RequestReply message:{}", message);
- }
+ LogUtils.info(log, "RequestReply message:{}", message);
final CloudEvent cloudEvent = EventMeshCloudEventBuilder.buildEventMeshCloudEvent(message, clientConfig, PROTOCOL_TYPE);
try {
final CloudEvent reply = publisherClient.withDeadlineAfter(timeout, TimeUnit.MILLISECONDS).requestReply(cloudEvent);
- if (log.isInfoEnabled()) {
- log.info("Received reply message:{}", reply);
- }
+ LogUtils.info(log, "Received reply message:{}", reply);
return EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(reply, PROTOCOL_TYPE);
} catch (Exception e) {
log.error("Error in RequestReply message {}", message, e);
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/EventMeshHttpConsumer.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/EventMeshHttpConsumer.java
index 24632a45b..d8cab6452 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/EventMeshHttpConsumer.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/EventMeshHttpConsumer.java
@@ -37,6 +37,7 @@ import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import java.util.ArrayList;
import java.util.Collections;
@@ -65,9 +66,8 @@ public class EventMeshHttpConsumer extends AbstractHttpClient implements AutoClo
this(eventMeshHttpClientConfig, null);
}
- public EventMeshHttpConsumer(final EventMeshHttpClientConfig eventMeshHttpClientConfig,
- final ThreadPoolExecutor customExecutor)
- throws EventMeshException {
+ public EventMeshHttpConsumer(final EventMeshHttpClientConfig eventMeshHttpClientConfig, final ThreadPoolExecutor customExecutor)
+ throws EventMeshException {
super(eventMeshHttpClientConfig);
this.consumeExecutor = Optional.ofNullable(customExecutor).orElseGet(
() -> ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpClientConfig.getConsumeThreadCore(),
@@ -167,9 +167,7 @@ public class EventMeshHttpConsumer extends AbstractHttpClient implements AutoClo
@Override
public void close() throws EventMeshException {
- if (log.isInfoEnabled()) {
- log.info("LiteConsumer shutdown begin.");
- }
+ LogUtils.info(log, "LiteConsumer shutdown begin.");
super.close();
if (consumeExecutor != null) {
@@ -177,9 +175,7 @@ public class EventMeshHttpConsumer extends AbstractHttpClient implements AutoClo
}
scheduler.shutdown();
- if (log.isInfoEnabled()) {
- log.info("LiteConsumer shutdown end.");
- }
+ LogUtils.info(log, "LiteConsumer shutdown end.");
}
private RequestParam buildCommonRequestParam() {
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/util/HttpUtils.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/util/HttpUtils.java
index b98c98a42..ba349f8e5 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/util/HttpUtils.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/util/HttpUtils.java
@@ -19,6 +19,7 @@ package org.apache.eventmesh.client.http.util;
import org.apache.eventmesh.client.http.model.RequestParam;
import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
@@ -108,9 +109,7 @@ public final class HttpUtils {
httpPost.setConfig(configBuilder.build());
- if (log.isDebugEnabled()) {
- log.debug("{}", httpPost);
- }
+ LogUtils.debug(log, "{}", httpPost);
return client.execute(httpPost, responseHandler);
}
@@ -162,9 +161,7 @@ public final class HttpUtils {
httpGet.setConfig(configBuilder.build());
- if (log.isDebugEnabled()) {
- log.debug("{}", httpGet);
- }
+ LogUtils.debug(log, "{}", httpGet);
return client.execute(httpGet, responseHandler);
}
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/RequestContext.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/RequestContext.java
index 0c5624116..8ee662d4b 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/RequestContext.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/RequestContext.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.client.tcp.common;
import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.utils.LogUtils;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -72,9 +73,7 @@ public class RequestContext {
public static RequestContext context(final Object key, final Package request) throws Exception {
final RequestContext context = new RequestContext(key, request);
- if (log.isInfoEnabled()) {
- log.info("_RequestContext|create|key={}", key);
- }
+ LogUtils.info(log, "_RequestContext|create|key={}", key);
return context;
}
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java
index da724327a..c42d9af27 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
import org.apache.eventmesh.common.protocol.tcp.codec.Codec;
+import org.apache.eventmesh.common.utils.LogUtils;
import java.io.Closeable;
import java.net.InetSocketAddress;
@@ -117,10 +118,8 @@ public abstract class TcpClient implements Closeable {
ChannelFuture f = bootstrap.connect(host, port).sync();
InetSocketAddress localAddress = (InetSocketAddress) f.channel().localAddress();
channel = f.channel();
- if (log.isInfoEnabled()) {
- log.info("connected|local={}:{}|server={}", localAddress.getAddress().getHostAddress(),
- localAddress.getPort(), host + ":" + port);
- }
+ LogUtils.info(log, "connected|local={}:{}|server={}", localAddress.getAddress().getHostAddress(),
+ localAddress.getPort(), host + ":" + port);
}
@Override
@@ -135,9 +134,7 @@ public abstract class TcpClient implements Closeable {
} catch (Exception e) {
Thread.currentThread().interrupt();
- if (log.isWarnEnabled()) {
- log.warn("close tcp client failed.|remote address={}", channel.remoteAddress(), e);
- }
+ LogUtils.warn(log, "close tcp client failed.|remote address={}", channel.remoteAddress(), e);
}
}
@@ -151,9 +148,7 @@ public abstract class TcpClient implements Closeable {
}
Package msg = MessageUtils.heartBeat();
io(msg, EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
- if (log.isDebugEnabled()) {
- log.debug("heart beat start {}", msg);
- }
+ LogUtils.debug(log, "heart beat start {}", msg);
} catch (Exception e) {
// ignore
}
@@ -175,9 +170,7 @@ public abstract class TcpClient implements Closeable {
if (channel.isWritable()) {
channel.writeAndFlush(msg).addListener((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
- if (log.isWarnEnabled()) {
- log.warn("send msg failed", future.cause());
- }
+ LogUtils.warn(log, "send msg failed", future.cause());
}
});
} else {
@@ -191,9 +184,7 @@ public abstract class TcpClient implements Closeable {
if (!contexts.containsValue(context)) {
contexts.put(key, context);
} else {
- if (log.isInfoEnabled()) {
- log.info("duplicate key : {}", key);
- }
+ LogUtils.info(log, "duplicate key : {}", key);
}
send(msg);
Supplier<Package> supplier = () -> {
@@ -223,10 +214,8 @@ public abstract class TcpClient implements Closeable {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- if (log.isInfoEnabled()) {
- log.info("exceptionCaught, close connection.|remote address={}",
- ctx.channel().remoteAddress(), cause);
- }
+ LogUtils.info(log, "exceptionCaught, close connection.|remote address={}",
+ ctx.channel().remoteAddress(), cause);
ctx.close();
}
};
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java
index f60e9e2ad..1f45d6ff2 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/impl/eventmeshmessage/EventMeshMessageTCPPubClient.java
@@ -31,6 +31,7 @@ import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import java.util.concurrent.ConcurrentHashMap;
@@ -79,9 +80,7 @@ class EventMeshMessageTCPPubClient extends TcpClient implements EventMeshTCPPubC
public Package rr(EventMeshMessage eventMeshMessage, long timeout) throws EventMeshException {
try {
Package msg = MessageUtils.buildPackage(eventMeshMessage, Command.REQUEST_TO_SERVER);
- if (log.isInfoEnabled()) {
- log.info("{}|rr|send|type={}|msg={}", CLIENTNO, msg, msg);
- }
+ LogUtils.info(log, "{}|rr|send|type={}|msg={}", CLIENTNO, msg, msg);
return io(msg, timeout);
} catch (Exception e) {
throw new EventMeshException("rr error", e);
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/workflow/EventMeshWorkflowClient.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/workflow/EventMeshWorkflowClient.java
index 8f8f4602d..212921273 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/workflow/EventMeshWorkflowClient.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/workflow/EventMeshWorkflowClient.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.client.workflow.config.EventMeshWorkflowClientConfig
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteRequest;
import org.apache.eventmesh.common.protocol.workflow.protos.ExecuteResponse;
import org.apache.eventmesh.common.protocol.workflow.protos.WorkflowGrpc;
+import org.apache.eventmesh.common.utils.LogUtils;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
@@ -57,9 +58,7 @@ public class EventMeshWorkflowClient {
public ExecuteResponse execute(final ExecuteRequest request) throws Exception {
final WorkflowGrpc.WorkflowBlockingStub workflowClient = getWorkflowClient();
final ExecuteResponse response = workflowClient.execute(request);
- if (log.isInfoEnabled()) {
- log.info("received response:{}", response);
- }
+ LogUtils.info(log, "received response:{}", response);
return response;
}
}
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/http/demo/SyncRequestInstance.java b/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/http/demo/SyncRequestInstance.java
index 619009ca6..fa298f99e 100644
--- a/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/http/demo/SyncRequestInstance.java
+++ b/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/http/demo/SyncRequestInstance.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
import org.apache.eventmesh.client.http.producer.EventMeshHttpProducer;
import org.apache.eventmesh.common.EventMeshMessage;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
@@ -65,10 +66,8 @@ public class SyncRequestInstance {
.uniqueId(RandomStringUtils.generateNum(30)).build();
EventMeshMessage rsp = eventMeshHttpProducer.request(eventMeshMessage, 10000);
- if (log.isDebugEnabled()) {
- log.debug("sendmsg : {}, return : {}, cost:{}ms", eventMeshMessage.getContent(), rsp.getContent(),
- System.currentTimeMillis() - startTime);
- }
+ LogUtils.debug(log, "sendmsg : {}, return : {}, cost:{}ms", eventMeshMessage.getContent(), rsp.getContent(),
+ System.currentTimeMillis() - startTime);
} catch (Exception e) {
log.warn("send msg failed", e);
}
diff --git a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManager.java b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManager.java
index af7e19c84..dc6783783 100644
--- a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManager.java
+++ b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManager.java
@@ -21,6 +21,7 @@ import static org.apache.eventmesh.webhook.api.WebHookOperationConstant.OPERATIO
import static org.apache.eventmesh.webhook.api.WebHookOperationConstant.OPERATION_MODE_NACOS;
import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.webhook.api.WebHookConfigOperation;
import org.apache.eventmesh.webhook.config.AdminConfiguration;
@@ -71,9 +72,8 @@ public class AdminWebHookConfigOperationManager {
try {
constructor.setAccessible(true);
final Properties operationProperties = adminConfiguration.getOperationProperties();
- if (log.isInfoEnabled()) {
- log.info("operationMode is {} properties is {} ", operationMode, operationProperties);
- }
+
+ LogUtils.info(log, "operationMode is {} properties is {} ", operationMode, operationProperties);
this.webHookConfigOperation = constructor.newInstance(operationProperties);
} finally {
// Restore the original accessibility of constructor
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 db222905c..b7b0f4945 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
@@ -18,6 +18,7 @@
package org.apache.eventmesh.webhook.admin;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.webhook.api.WebHookConfig;
import org.apache.eventmesh.webhook.api.WebHookConfigOperation;
import org.apache.eventmesh.webhook.api.WebHookOperationConstant;
@@ -62,10 +63,8 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
@Override
public Integer insertWebHookConfig(final WebHookConfig webHookConfig) {
if (!webHookConfig.getCallbackPath().startsWith(WebHookOperationConstant.CALLBACK_PATH_PREFIX)) {
- if (log.isErrorEnabled()) {
- log.error("webhookConfig callback path must start with {}",
- WebHookOperationConstant.CALLBACK_PATH_PREFIX);
- }
+ LogUtils.error(log, "webhookConfig callback path must start with {}",
+ WebHookOperationConstant.CALLBACK_PATH_PREFIX);
return 0;
}
@@ -76,9 +75,7 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
if (webhookConfigFile.exists()) {
- if (log.isErrorEnabled()) {
- log.error("webhookConfig {} exists", webHookConfig.getCallbackPath());
- }
+ LogUtils.error(log, "webhookConfig {} exists", webHookConfig.getCallbackPath());
return 0;
}
return writeToFile(webhookConfigFile, webHookConfig) ? 1 : 0;
@@ -88,9 +85,7 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
public Integer updateWebHookConfig(final WebHookConfig webHookConfig) {
final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
if (!webhookConfigFile.exists()) {
- if (log.isErrorEnabled()) {
- log.error("webhookConfig {} does not exist", webHookConfig.getCallbackPath());
- }
+ LogUtils.error(log, "webhookConfig {} does not exist", webHookConfig.getCallbackPath());
return 0;
}
return writeToFile(webhookConfigFile, webHookConfig) ? 1 : 0;
@@ -101,9 +96,7 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
synchronized (SharedLatchHolder.lock) {
final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
if (!webhookConfigFile.exists()) {
- if (log.isErrorEnabled()) {
- log.error("webhookConfig {} does not exist", webHookConfig.getCallbackPath());
- }
+ LogUtils.error(log, "webhookConfig {} does not exist", webHookConfig.getCallbackPath());
return 0;
}
return webhookConfigFile.delete() ? 1 : 0;
@@ -117,9 +110,7 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
public WebHookConfig queryWebHookConfigById(final WebHookConfig webHookConfig) {
final File webhookConfigFile = getWebhookConfigFile(webHookConfig);
if (!webhookConfigFile.exists()) {
- if (log.isErrorEnabled()) {
- log.error("webhookConfig {} does not exist", webHookConfig.getCallbackPath());
- }
+ LogUtils.error(log, "webhookConfig {} does not exist", webHookConfig.getCallbackPath());
return null;
}
@@ -133,9 +124,7 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
final String manuDirPath = getWebhookConfigManuDir(webHookConfig);
final File manuDir = new File(manuDirPath);
if (!manuDir.exists()) {
- if (log.isWarnEnabled()) {
- log.warn("webhookConfig dir {} does not exist", manuDirPath);
- }
+ LogUtils.warn(log, "webhookConfig dir {} does not exist", manuDirPath);
return new ArrayList<>();
}
@@ -166,9 +155,7 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
fileContent.append(line);
}
} catch (IOException e) {
- if (log.isErrorEnabled()) {
- log.error("get webHookConfig from file {} error", webhookConfigFile.getPath(), e);
- }
+ LogUtils.error(log, "get webHookConfig from file {} error", webhookConfigFile.getPath(), e);
return null;
}
@@ -184,9 +171,7 @@ public class FileWebHookConfigOperation implements WebHookConfigOperation {
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());
- }
+ LogUtils.error(log, "write webhookConfig {} to file error", webHookConfig.getCallbackPath());
return false;
}
return true;
diff --git a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/NacosWebHookConfigOperation.java b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/NacosWebHookConfigOperation.java
index 495b9ae79..98e43a8e8 100644
--- a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/NacosWebHookConfigOperation.java
+++ b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/NacosWebHookConfigOperation.java
@@ -23,6 +23,7 @@ import static org.apache.eventmesh.webhook.api.WebHookOperationConstant.MANUFACT
import static org.apache.eventmesh.webhook.api.WebHookOperationConstant.TIMEOUT_MS;
import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.webhook.api.Manufacturer;
import org.apache.eventmesh.webhook.api.WebHookConfig;
import org.apache.eventmesh.webhook.api.WebHookConfigOperation;
@@ -61,10 +62,8 @@ public class NacosWebHookConfigOperation implements WebHookConfigOperation {
@Override
public Integer insertWebHookConfig(final WebHookConfig webHookConfig) {
if (!webHookConfig.getCallbackPath().startsWith(WebHookOperationConstant.CALLBACK_PATH_PREFIX)) {
- if (log.isErrorEnabled()) {
- log.error("webhookConfig callback path must start with {}",
- WebHookOperationConstant.CALLBACK_PATH_PREFIX);
- }
+ LogUtils.error(log, "webhookConfig callback path must start with {}",
+ WebHookOperationConstant.CALLBACK_PATH_PREFIX);
return 0;
}
@@ -73,9 +72,7 @@ public class NacosWebHookConfigOperation implements WebHookConfigOperation {
try {
if (configService.getConfig(getWebHookConfigDataId(webHookConfig),
getManuGroupId(webHookConfig), TIMEOUT_MS) != null) {
- if (log.isErrorEnabled()) {
- log.error("insertWebHookConfig failed, config has existed");
- }
+ LogUtils.error(log, "insertWebHookConfig failed, config has existed");
return 0;
}
result = configService.publishConfig(getWebHookConfigDataId(webHookConfig), getManuGroupId(webHookConfig),
@@ -112,9 +109,7 @@ public class NacosWebHookConfigOperation implements WebHookConfigOperation {
try {
if (configService.getConfig(getWebHookConfigDataId(webHookConfig), getManuGroupId(webHookConfig),
TIMEOUT_MS) == null) {
- if (log.isErrorEnabled()) {
- log.error("updateWebHookConfig failed, config is not existed");
- }
+ LogUtils.error(log, "updateWebHookConfig failed, config is not existed");
return 0;
}
result = configService.publishConfig(getWebHookConfigDataId(webHookConfig),
diff --git a/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java b/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java
index 8884b1dce..b25e13f4b 100644
--- a/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java
+++ b/eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.WebhookProtocolTransportObject;
+import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
import org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
import org.apache.eventmesh.webhook.api.WebHookConfig;
@@ -122,16 +123,12 @@ public class WebHookController {
@Override
public void onSuccess(SendResult sendResult) {
- if (log.isDebugEnabled()) {
- log.debug(sendResult.toString());
- }
+ LogUtils.debug(log, sendResult.toString());
}
@Override
public void onException(OnExceptionContext context) {
- if (log.isWarnEnabled()) {
- log.warn("", context.getException());
- }
+ LogUtils.warn(log, "", context.getException());
}
});
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org