[ISSUE #4646] Add condition for transforming non-data values

eventmesh-connectors/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/server/LarkConnectServer.java

@@ -28,7 +28,7 @@ public class LarkConnectServer {
     public static void main(String[] args) throws Exception {
 
         LarkConnectServerConfig larkConnectServerConfig = ConfigUtil.parse(LarkConnectServerConfig.class,
-                Constants.CONNECT_SERVER_CONFIG_FILE_NAME);
+            Constants.CONNECT_SERVER_CONFIG_FILE_NAME);
 
         if (larkConnectServerConfig.isSinkEnable()) {
             Application application = new Application();

eventmesh-connectors/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/sink/ImServiceHandler.java

@@ -93,14 +93,13 @@
         ImServiceHandler imServiceHandler = new ImServiceHandler();
         imServiceHandler.sinkConnectorConfig = sinkConnectorConfig;
         imServiceHandler.imService = Client.newBuilder(sinkConnectorConfig.getAppId(), sinkConnectorConfig.getAppSecret())
-                .httpTransport(new OkHttpTransport(new OkHttpClient().newBuilder()
-                        .callTimeout(3L, TimeUnit.SECONDS)
-                        .build())
-                )
-                .disableTokenCache()
-                .requestTimeout(3, TimeUnit.SECONDS)
-                .build()
-                .im();
+            .httpTransport(new OkHttpTransport(new OkHttpClient().newBuilder()
+                .callTimeout(3L, TimeUnit.SECONDS)
+                .build()))
+            .disableTokenCache()
+            .requestTimeout(3, TimeUnit.SECONDS)
+            .build()
+            .im();
 
         long fixedWait = Long.parseLong(sinkConnectorConfig.getRetryDelayInMills());
         int maxRetryTimes = Integer.parseInt(sinkConnectorConfig.getMaxRetryTimes()) + 1;
@@ -128,24 +127,25 @@
             });
         } else {
             imServiceHandler.retryer = RetryerBuilder.<ConnectRecord>newBuilder()
-                    .retryIfException()
-                    .retryIfResult(Objects::nonNull)
-                    .withWaitStrategy(WaitStrategies.fixedWait(fixedWait, TimeUnit.MILLISECONDS))
-                    .withStopStrategy(StopStrategies.stopAfterAttempt(maxRetryTimes))
-                    .withRetryListener(new RetryListener() {
-                        @SneakyThrows
-                        @Override
-                        public <V> void onRetry(Attempt<V> attempt) {
-
-                            long times = attempt.getAttemptNumber();
-                            if (times > 1) {
-                                redoSinkNum.increment();
-                                log.info("Total redo sink task num : [{}]", redoSinkNum.sum());
-                                log.warn("Retry sink event to lark | times=[{}]", attempt.getAttemptNumber() - 1);
-                            }
+                .retryIfException()
+                .retryIfResult(Objects::nonNull)
+                .withWaitStrategy(WaitStrategies.fixedWait(fixedWait, TimeUnit.MILLISECONDS))
+                .withStopStrategy(StopStrategies.stopAfterAttempt(maxRetryTimes))
+                .withRetryListener(new RetryListener() {
+
+                    @SneakyThrows
+                    @Override
+                    public <V> void onRetry(Attempt<V> attempt) {
+
+                        long times = attempt.getAttemptNumber();
+                        if (times > 1) {
+                            redoSinkNum.increment();
+                            log.info("Total redo sink task num : [{}]", redoSinkNum.sum());
+                            log.warn("Retry sink event to lark | times=[{}]", attempt.getAttemptNumber() - 1);
                         }
-                    })
-                    .build();
+                    }
+                })
+                .build();
         }
 
         return imServiceHandler;
@@ -156,9 +156,9 @@
         headers.put("Content-Type", Lists.newArrayList("application/json; charset=utf-8"));
 
         RequestOptions requestOptions = RequestOptions.newBuilder()
-                .tenantAccessToken(getTenantAccessToken(sinkConnectorConfig.getAppId(), sinkConnectorConfig.getAppSecret()))
-                .headers(headers)
-                .build();
+            .tenantAccessToken(getTenantAccessToken(sinkConnectorConfig.getAppId(), sinkConnectorConfig.getAppSecret()))
+            .headers(headers)
+            .build();
 
         retryer.call(() -> {
             CreateMessageReq createMessageReq = convertCreateMessageReq(connectRecord);
@@ -176,9 +176,9 @@
         headers.put("Content-Type", Lists.newArrayList("application/json; charset=utf-8"));
 
         RequestOptions requestOptions = RequestOptions.newBuilder()
-                .tenantAccessToken(getTenantAccessToken(sinkConnectorConfig.getAppId(), sinkConnectorConfig.getAppSecret()))
-                .headers(headers)
-                .build();
+            .tenantAccessToken(getTenantAccessToken(sinkConnectorConfig.getAppId(), sinkConnectorConfig.getAppSecret()))
+            .headers(headers)
+            .build();
 
         CreateMessageReq createMessageReq = convertCreateMessageReq(connectRecord);
 
@@ -187,30 +187,30 @@
         LongAdder cnt = new LongAdder();
         AtomicBoolean isAck = new AtomicBoolean(false);
         Runnable task = () -> CompletableFuture
-                .supplyAsync(() -> {
-                    try {
-                        cnt.increment();
-                        return imService.message().create(createMessageReq, requestOptions);
-                    } catch (Exception e) {
-                        throw new RuntimeException(e);
-                    }
-                }, sinkAsyncWorker)
-                .whenCompleteAsync((resp, e) -> {
-                    if (cnt.sum() > 1) {
-                        redoSinkNum.increment();
-                        log.info("Total redo sink task num : [{}]", redoSinkNum.sum());
-                        log.warn("Retry sink event to lark | times=[{}]", cnt.sum() - 1);
-                    }
-                    if (Objects.nonNull(e)) {
-                        log.error("eventmesh-connector-lark internal exception.", e);
-                        return;
-                    }
-                    if (resp.getCode() != 0) {
-                        log.warn("Sinking event to lark failure | code:[{}] | msg:[{}] | err:[{}]", resp.getCode(), resp.getMsg(), resp.getError());
-                        return;
-                    }
-                    isAck.set(true);
-                });
+            .supplyAsync(() -> {
+                try {
+                    cnt.increment();
+                    return imService.message().create(createMessageReq, requestOptions);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }, sinkAsyncWorker)
+            .whenCompleteAsync((resp, e) -> {
+                if (cnt.sum() > 1) {
+                    redoSinkNum.increment();
+                    log.info("Total redo sink task num : [{}]", redoSinkNum.sum());
+                    log.warn("Retry sink event to lark | times=[{}]", cnt.sum() - 1);
+                }
+                if (Objects.nonNull(e)) {
+                    log.error("eventmesh-connector-lark internal exception.", e);
+                    return;
+                }
+                if (resp.getCode() != 0) {
+                    log.warn("Sinking event to lark failure | code:[{}] | msg:[{}] | err:[{}]", resp.getCode(), resp.getMsg(), resp.getError());
+                    return;
+                }
+                isAck.set(true);
+            });
 
         ScheduledFuture<?> future = retryWorker.scheduleAtFixedRate(task, 0L, fixedWait, TimeUnit.MILLISECONDS);
         cleanerWorker.submit(() -> {
@@ -226,8 +226,8 @@
 
     private CreateMessageReq convertCreateMessageReq(ConnectRecord connectRecord) {
         CreateMessageReqBody.Builder bodyBuilder = CreateMessageReqBody.newBuilder()
-                .receiveId(sinkConnectorConfig.getReceiveId())
-                .uuid(UUID.randomUUID().toString());
+            .receiveId(sinkConnectorConfig.getReceiveId())
+            .uuid(UUID.randomUUID().toString());
 
         String templateTypeKey = connectRecord.getExtension(ConnectRecordExtensionKeys.TEMPLATE_TYPE_4_LARK);
         if (null == templateTypeKey || "null".equals(templateTypeKey)) {
@@ -236,18 +236,18 @@
         LarkMessageTemplateType templateType = LarkMessageTemplateType.of(templateTypeKey);
         if (LarkMessageTemplateType.PLAIN_TEXT == templateType) {
             bodyBuilder.content(createTextContent(connectRecord))
-                    .msgType(MsgTypeEnum.MSG_TYPE_TEXT.getValue());
+                .msgType(MsgTypeEnum.MSG_TYPE_TEXT.getValue());
         } else if (LarkMessageTemplateType.MARKDOWN == templateType) {
             String title = Optional.ofNullable(connectRecord.getExtension(ConnectRecordExtensionKeys.MARKDOWN_MESSAGE_TITLE_4_LARK))
-                    .orElse("EventMesh-Message");
+                .orElse("EventMesh-Message");
             bodyBuilder.content(createInteractiveContent(connectRecord, title))
-                    .msgType(MsgTypeEnum.MSG_TYPE_INTERACTIVE.getValue());
+                .msgType(MsgTypeEnum.MSG_TYPE_INTERACTIVE.getValue());
         }
 
         return CreateMessageReq.newBuilder()
-                .receiveIdType(sinkConnectorConfig.getReceiveIdType())
-                .createMessageReqBody(bodyBuilder.build())
-                .build();
+            .receiveIdType(sinkConnectorConfig.getReceiveIdType())
+            .createMessageReqBody(bodyBuilder.build())
+            .build();
     }
 
     private String createTextContent(ConnectRecord connectRecord) {
@@ -287,26 +287,26 @@
         sb.append(new String((byte[]) connectRecord.getData()));
 
         MessageCardConfig config = MessageCardConfig.newBuilder()
-                .enableForward(true)
-                .wideScreenMode(true)
-                .updateMulti(true)
-                .build();
+            .enableForward(true)
+            .wideScreenMode(true)
+            .updateMulti(true)
+            .build();
 
         // header
         MessageCardHeader header = MessageCardHeader.newBuilder()
-                .template(MessageCardHeaderTemplateEnum.BLUE)
-                .title(MessageCardPlainText.newBuilder()
-                        .content(title)
-                        .build())
-                .build();
+            .template(MessageCardHeaderTemplateEnum.BLUE)
+            .title(MessageCardPlainText.newBuilder()
+                .content(title)
+                .build())
+            .build();
 
         MessageCard content = MessageCard.newBuilder()
-                .config(config)
-                .header(header)
-                .elements(new MessageCardElement[]{
-                        MessageCardMarkdown.newBuilder().content(sb.toString()).build()
-                })
-                .build();
+            .config(config)
+            .header(header)
+            .elements(new MessageCardElement[]{
+                    MessageCardMarkdown.newBuilder().content(sb.toString()).build()
+            })
+            .build();
 
         return content.String();
     }
@@ -328,7 +328,7 @@
      */
     private void atAll(StringBuilder sb) {
         sb.append("<at id=all>")
-                .append("</at>");
+            .append("</at>");
     }
 
     /**
@@ -339,8 +339,8 @@
      */
     private void atUser(StringBuilder sb, String userId) {
         sb.append("<at id=")
-                .append(userId)
-                .append(">")
-                .append("</at>");
+            .append(userId)
+            .append(">")
+            .append("</at>");
     }
 }

eventmesh-connectors/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/sink/config/SinkConnectorConfig.java

@@ -70,12 +70,12 @@
 
         // validate receiveIdType
         if (!StringUtils.containsAny(receiveIdType, ReceiveIdTypeEnum.CHAT_ID.getValue(),
-                ReceiveIdTypeEnum.EMAIL.getValue(),
-                ReceiveIdTypeEnum.OPEN_ID.getValue(),
-                ReceiveIdTypeEnum.USER_ID.getValue(),
-                ReceiveIdTypeEnum.UNION_ID.getValue())) {
+            ReceiveIdTypeEnum.EMAIL.getValue(),
+            ReceiveIdTypeEnum.OPEN_ID.getValue(),
+            ReceiveIdTypeEnum.USER_ID.getValue(),
+            ReceiveIdTypeEnum.UNION_ID.getValue())) {
             throw new IllegalArgumentException(
-                    String.format("sinkConnectorConfig.receiveIdType=[%s], Invalid.", receiveIdType));
+                String.format("sinkConnectorConfig.receiveIdType=[%s], Invalid.", receiveIdType));
         }
     }
 }

eventmesh-connectors/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/sink/connector/LarkSinkConnector.java

@@ -44,7 +44,6 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
-
 @Slf4j
 public class LarkSinkConnector implements Sink {
 
@@ -60,11 +59,11 @@
      * you can try to modify it.
      */
     public static final Cache<String, String> AUTH_CACHE = CacheBuilder.newBuilder()
-            .initialCapacity(12)
-            .maximumSize(10)
-            .concurrencyLevel(5)
-            .expireAfterWrite(30, TimeUnit.MINUTES)
-            .build();
+        .initialCapacity(12)
+        .maximumSize(10)
+        .concurrencyLevel(5)
+        .expireAfterWrite(30, TimeUnit.MINUTES)
+        .build();
 
     private LarkSinkConfig sinkConfig;
 
@@ -78,7 +77,8 @@
     }
 
     @Override
-    public void init(Config config) {}
+    public void init(Config config) {
+    }
 
     @Override
     public void init(ConnectorContext connectorContext) {
@@ -136,15 +136,15 @@
         return AUTH_CACHE.get(TENANT_ACCESS_TOKEN, () -> {
 
             Client client = Client.newBuilder(appId, appSecret)
-                    .appType(AppType.SELF_BUILT)
-                    .logReqAtDebug(true)
-                    .build();
+                .appType(AppType.SELF_BUILT)
+                .logReqAtDebug(true)
+                .build();
 
             TenantAccessTokenResp resp = client.ext().getTenantAccessTokenBySelfBuiltApp(
-                    SelfBuiltTenantAccessTokenReq.newBuilder()
-                            .appSecret(appSecret)
-                            .appId(appId)
-                            .build());
+                SelfBuiltTenantAccessTokenReq.newBuilder()
+                    .appSecret(appSecret)
+                    .appId(appId)
+                    .build());
             return resp.getTenantAccessToken();
         });
     }

eventmesh-connectors/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/sink/ImServiceHandlerTest.java

@@ -80,8 +80,8 @@ private void init() throws Exception {
         when(message.create(any(), any())).thenReturn(new CreateMessageResp());
         when(imService.message()).thenReturn(message);
         Field imServiceField = ReflectionSupport.findFields(imServiceHandler.getClass(),
-                (f) -> f.getName().equals("imService"),
-                HierarchyTraversalMode.BOTTOM_UP).get(0);
+            (f) -> f.getName().equals("imService"),
+            HierarchyTraversalMode.BOTTOM_UP).get(0);
         imServiceField.setAccessible(true);
         imServiceField.set(imServiceHandler, imService);
     }
@@ -106,7 +106,7 @@ private void regularSink() throws Exception {
             RecordPartition partition = new RecordPartition();
             RecordOffset offset = new RecordOffset();
             ConnectRecord connectRecord = new ConnectRecord(partition, offset,
-                    System.currentTimeMillis(), "test-lark".getBytes(StandardCharsets.UTF_8));
+                System.currentTimeMillis(), "test-lark".getBytes(StandardCharsets.UTF_8));
             if (Boolean.parseBoolean(sinkConnectorConfig.getSinkAsync())) {
                 imServiceHandler.sinkAsync(connectRecord);
                 long retryDelayInMills = Long.parseLong(sinkConnectorConfig.getRetryDelayInMills());
@@ -148,7 +148,7 @@ private void retrySink() throws Exception {
             RecordPartition partition = new RecordPartition();
             RecordOffset offset = new RecordOffset();
             ConnectRecord connectRecord = new ConnectRecord(partition, offset,
-                    System.currentTimeMillis(), "test-lark".getBytes(StandardCharsets.UTF_8));
+                System.currentTimeMillis(), "test-lark".getBytes(StandardCharsets.UTF_8));
             if (Boolean.parseBoolean(sinkConnectorConfig.getSinkAsync())) {
                 imServiceHandler.sinkAsync(connectRecord);
 

eventmesh-connectors/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/sink/LarkSinkConnectorTest.java

@@ -85,7 +85,7 @@ public void testPut() throws Exception {
             RecordPartition partition = new RecordPartition();
             RecordOffset offset = new RecordOffset();
             ConnectRecord connectRecord = new ConnectRecord(partition, offset,
-                    System.currentTimeMillis(), "test-lark".getBytes(StandardCharsets.UTF_8));
+                System.currentTimeMillis(), "test-lark".getBytes(StandardCharsets.UTF_8));
             connectRecords.add(connectRecord);
         }
         larkSinkConnector.put(connectRecords);

eventmesh-connectors/eventmesh-connector-wechat/src/test/java/org/apache/eventmesh/connector/wechat/sink/connector/WeChatSinkConnectorTest.java

@@ -55,8 +55,6 @@
 import okhttp3.Response;
 import okhttp3.ResponseBody;
 
-
-
 @ExtendWith(MockitoExtension.class)
 public class WeChatSinkConnectorTest {
 
@@ -106,8 +104,8 @@ public void testSendMessageToWeChat() throws Exception {
             .body(sendMessageBody)
             .message("ok")
             .build();
-        ArgumentMatcher<Request> sendMessageMatcher = (anyRequest) ->
-            sendMessageRequest.url().encodedPath().startsWith(anyRequest.url().encodedPath());
+        ArgumentMatcher<Request> sendMessageMatcher =
+            (anyRequest) -> sendMessageRequest.url().encodedPath().startsWith(anyRequest.url().encodedPath());
         Call sendMessageRequestCall = Mockito.mock(Call.class);
         Mockito.doReturn(sendMessageRequestCall).when(okHttpClient).newCall(Mockito.argThat(sendMessageMatcher));
         Mockito.doReturn(sendMessageResponse).when(sendMessageRequestCall).execute();
@@ -137,8 +135,8 @@ public void testSendMessageToWeChatAbnormally() throws Exception {
             .body(sendMessageBody)
             .message("ok")
             .build();
-        ArgumentMatcher<Request> sendMessageMatcher = (anyRequest) ->
-            sendMessageRequest.url().encodedPath().startsWith(anyRequest.url().encodedPath());
+        ArgumentMatcher<Request> sendMessageMatcher =
+            (anyRequest) -> sendMessageRequest.url().encodedPath().startsWith(anyRequest.url().encodedPath());
         Call sendMessageRequestCall = Mockito.mock(Call.class);
         Mockito.doReturn(sendMessageRequestCall).when(okHttpClient).newCall(Mockito.argThat(sendMessageMatcher));
         Mockito.doReturn(sendMessageResponse).when(sendMessageRequestCall).execute();