[GitHub] [inlong] leezng opened a new pull request, #5902: [INLONG-5901][Dashboard] Unified sinks type definition
leezng opened a new pull request, #5902: URL: https://github.com/apache/inlong/pull/5902 - Fixes #5901 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang commented on a diff in pull request #5852: [INLONG-5851][TubeMQ] Optimize while-sleep to ScheduledExecutorService
dockerzhang commented on code in PR #5852: URL: https://github.com/apache/inlong/pull/5852#discussion_r971740319 ## inlong-tubemq/tubemq-manager/pom.xml: ## @@ -176,6 +176,11 @@ jakarta.validation jakarta.validation-api + +io.netty +netty-common + 4.1.77.Final Review Comment: version is not needed here. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang commented on pull request #5852: [INLONG-5851][TubeMQ] Optimize while-sleep to ScheduledExecutorService
dockerzhang commented on PR #5852: URL: https://github.com/apache/inlong/pull/5852#issuecomment-1247829521 please remove the LICENSE-netty.txt and NOTICE-netty.txt file. for apache license dependency, you can add it to LICENSE file directly. you can refer to the [netty-common](https://github.com/apache/inlong/blob/master/licenses/inlong-manager/LICENSE#L505) in inlong-manager. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #5902: [INLONG-5901][Dashboard] Unified sinks type definition
dockerzhang merged PR #5902: URL: https://github.com/apache/inlong/pull/5902 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] ly100120 commented on issue #5110: [Bug][DataProxy] DataProxy is not sending data to Pulsar
ly100120 commented on issue #5110: URL: https://github.com/apache/inlong/issues/5110#issuecomment-1248013888 这个问题你是否已经解决？我碰到了同样的问题，束手无策。1.4.0版本 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] haifxu opened a new pull request, #5905: [INLONG-5904][Manager] Add manager client packages into lib directory
haifxu opened a new pull request, #5905: URL: https://github.com/apache/inlong/pull/5905 ### Prepare a Pull Request - Fixes #5904 ### Motivation  ### Modifications 1.Add packaging configuration. 2.Add them into lib directory. ### Verifying this change - [ ] This change is a trivial rework/code cleanup without any test coverage. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] thesumery opened a new pull request, #5906: [INLONG-5903][Sort] Make InLong metric constructs factory more cohesive
thesumery opened a new pull request, #5906: URL: https://github.com/apache/inlong/pull/5906 ### Prepare a Pull Request *(Change the title refer to the following example)* *[INLONG-5903][Sort] Make InLong metric constructs factory more cohesive* *(The following *XYZ* should be replaced by the actual [GitHub Issue](https://github.com/apache/inlong/issues) number)* - Fixes #5903 ### Motivation *To make inlong sort metric code logic more clearly * ### Modifications *Refactor `SinkMetricData` and `SourceMetricData` constuct function.* *Add MetricOption to replace scattered metric configuration* *Modify `inlong.audit` configuration to `metrics.audit.proxy.hosts` to keep consistent with manager.* -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] woofyzhao commented on pull request #5898: [INLONG-5897][Manager] Support query audit data by user role
woofyzhao commented on PR #5898: URL: https://github.com/apache/inlong/pull/5898#issuecomment-1248050074 > 1.Explain why users only pay attention to AUDIT_ID_DATAPROXY_RECEIVED and AUDIT_ID_SORT_OUTPUT? 2.It is suggested that querying audit views based on roles is only a default behavior. Users can also query any audit views according to their own needs. Excellent idea. Fixed as required. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5898: [INLONG-5897][Manager] Support query audit data by user role
healchow commented on code in PR #5898:
URL: https://github.com/apache/inlong/pull/5898#discussion_r971968846
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java:
##
@@ -70,12 +78,21 @@ public class AuditServiceImpl implements AuditService {
private static final String HOUR_FORMAT = "-MM-dd HH";
private static final String DAY_FORMAT = "-MM-dd";
+@Value("#{'${audit.admin.ids:3,4,5,6,7,8}'.split(',')}")
Review Comment:
1. The number is not clear for users.
2. How to set those ids? In a properties file?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] liangyepianzhou opened a new pull request, #5908: [INLONG-5907][TubeMQ] Replace the while-sleep with ScheduledExecutorService for tube server loopProcess
liangyepianzhou opened a new pull request, #5908: URL: https://github.com/apache/inlong/pull/5908 ### Prepare a Pull Request - Fixes https://github.com/apache/inlong/issues/5907 ### Motivation it is not recommended to execute a scheduled task by while-sleep. ### Modifications Replace while-sleep with ScheduledExecutorService. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] vernedeng opened a new issue, #560: [Feature][SortStandalone] Update SortStandalone docs
vernedeng opened a new issue, #560: URL: https://github.com/apache/inlong-website/issues/560 ### Description _No response_ ### Use case _No response_ ### Are you willing to submit PR? - [X] Yes, I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] vernedeng opened a new pull request, #561: [INLONG-560][Doc] Update sort-standalone docs
vernedeng opened a new pull request, #561: URL: https://github.com/apache/inlong-website/pull/561 ### Prepare a Pull Request *(Change the title refer to the following example)* - Title Example: [INLONG-XYZ][Component] Title of the pull request *(The following *XYZ* should be replaced by the actual [GitHub Issue](https://github.com/apache/inlong/issues) number)* - Fixes #560 ### Motivation *Explain here the context, and why you're making that change. What is the problem you're trying to solve.* ### Modifications *Describe the modifications you've done.* ### Verifying this change - [ ] Make sure that the change passes the CI checks. *(Please pick either of the following options)* This change is a trivial rework / code cleanup without any test coverage. *(or)* This change is already covered by existing tests, such as *(please describe tests)*. *(or)* This change added tests and can be verified as follows: *(example:)* - *Added integration tests for end-to-end deployment with large payloads (10MB)* - *Extended integration test for recovery after broker failure* ### Documentation - Does this pull request introduce a new feature? (yes / no) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) - If a feature is not applicable for documentation, explain why? - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] GanfengTan opened a new pull request, #5910: [INLONG-5909][Agent] Fix WatchService is no effect in the k8s
GanfengTan opened a new pull request, #5910: URL: https://github.com/apache/inlong/pull/5910 Fix WatchService is no effect in the k8s. - Fixes #5909 ### Motivation Trigger a new task when Data changes, for example: add data. ### Modifications Validate the job for same GroupId and StreamId. ### Verifying this change *(Please pick either of the following options)* - [x] This change is a trivial rework/code cleanup without any test coverage. - [ ] This change is already covered by existing tests, such as: *(please describe tests)* - [ ] This change added tests and can be verified as follows: -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #5900: [INLONG-5899][DataProxy] Optimize Http's SimpleMessageHandler class
dockerzhang merged PR #5900: URL: https://github.com/apache/inlong/pull/5900 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated: [INLONG-5899][DataProxy] Optimize Http's SimpleMessageHandler class (#5900)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new c5b9319a8 [INLONG-5899][DataProxy] Optimize Http's
SimpleMessageHandler class (#5900)
c5b9319a8 is described below
commit c5b9319a8fc49ebddad92a4f8193073db7e3f5fc
Author: Goson Zhang <4675...@qq.com>
AuthorDate: Fri Sep 16 10:07:32 2022 +0800
[INLONG-5899][DataProxy] Optimize Http's SimpleMessageHandler class (#5900)
---
.../dataproxy/http/SimpleMessageHandler.java | 135 +++--
.../inlong/dataproxy/utils/DateTimeUtils.java | 41 +++
2 files changed, 109 insertions(+), 67 deletions(-)
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/http/SimpleMessageHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/http/SimpleMessageHandler.java
index 1969f7cea..3f1bf425c 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/http/SimpleMessageHandler.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/http/SimpleMessageHandler.java
@@ -20,11 +20,6 @@ package org.apache.inlong.dataproxy.http;
import static
org.apache.inlong.dataproxy.consts.AttributeConstants.SEP_HASHTAG;
import javax.servlet.http.HttpServletRequest;
import java.io.UnsupportedEncodingException;
-import java.text.SimpleDateFormat;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
@@ -45,7 +40,7 @@ import
org.apache.inlong.dataproxy.metrics.DataProxyMetricItem;
import org.apache.inlong.dataproxy.metrics.DataProxyMetricItemSet;
import org.apache.inlong.dataproxy.metrics.audit.AuditUtils;
import org.apache.inlong.dataproxy.source.ServiceDecoder;
-import org.apache.inlong.dataproxy.utils.Constants;
+import org.apache.inlong.dataproxy.utils.DateTimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,9 +48,6 @@ public class SimpleMessageHandler implements MessageHandler {
private static final Logger LOG =
LoggerFactory.getLogger(SimpleMessageHandler.class);
private static final ConfigManager configManager =
ConfigManager.getInstance();
-private static final DateTimeFormatter DATE_FORMATTER
-= DateTimeFormatter.ofPattern("MMddHHmm");
-private static final ZoneId defZoneId = ZoneId.systemDefault();
private static final String DEFAULT_REMOTE_IDC_VALUE = "0";
private final MonitorIndex monitorIndex;
@@ -88,96 +80,105 @@ public class SimpleMessageHandler implements
MessageHandler {
@Override
public void processMessage(Context context) throws MessageProcessException
{
-String topicValue = "test";
-String attr = "m=0";
-StringBuilder newAttrBuffer = new StringBuilder(attr);
-
+StringBuilder strBuff = new StringBuilder(512);
+// get groupId and streamId
String groupId = (String) context.get(AttributeConstants.GROUP_ID);
String streamId = (String) context.get(AttributeConstants.STREAM_ID);
-String dt = (String) context.get(AttributeConstants.DATA_TIME);
-
-String value = getTopic(groupId, streamId);
-if (null != value && !"".equals(value)) {
-topicValue = value.trim();
+if (StringUtils.isBlank(groupId) || StringUtils.isBlank(streamId)) {
+throw new MessageProcessException(strBuff.append("Field ")
+.append(AttributeConstants.GROUP_ID).append(" or ")
+.append(AttributeConstants.STREAM_ID)
+.append(" must exist and not blank!").toString());
}
-
-String mxValue = configManager.getMxProperties().get(groupId);
-if (null != mxValue) {
-newAttrBuffer = new StringBuilder(mxValue.trim());
+groupId = groupId.trim();
+streamId = streamId.trim();
+// get topicName
+String topicName = "test";
+String configedTopicName = getTopic(groupId, streamId);
+if (StringUtils.isNotBlank(configedTopicName)) {
+topicName = configedTopicName.trim();
}
-
-
newAttrBuffer.append("&groupId=").append(groupId).append("&streamId=").append(streamId)
-.append("&dt=").append(dt);
+// get message data time
+final long msgRcvTime = System.currentTimeMillis();
+String strDataTime = (String)
context.get(AttributeConstants.DATA_TIME);
+long longDataTime = NumberUtils.toLong(strDataTime, msgRcvTime);
+strDataTime = String.valueOf(longDataTime);
+// get char set
+String charset = (String) context.get(AttributeConstants.CHARSET);
+if (StringUtils
[GitHub] [inlong] woofyzhao commented on a diff in pull request #5898: [INLONG-5897][Manager] Support query audit data by user role
woofyzhao commented on code in PR #5898:
URL: https://github.com/apache/inlong/pull/5898#discussion_r972549615
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java:
##
@@ -70,12 +78,21 @@ public class AuditServiceImpl implements AuditService {
private static final String HOUR_FORMAT = "-MM-dd HH";
private static final String DAY_FORMAT = "-MM-dd";
+@Value("#{'${audit.admin.ids:3,4,5,6,7,8}'.split(',')}")
Review Comment:
1. doc and inlong audit definition url link added.
2. set those ids in application.properties like:
audit.admin.ids = 3,4,5,6,7,8
audit.user.ids = 3,8
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] leezng opened a new pull request, #5912: [INLONG-5911][Dashboard] Modify the data group name to the data stream group
leezng opened a new pull request, #5912: URL: https://github.com/apache/inlong/pull/5912 - Fixes #5911 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] vernedeng opened a new issue, #562: [Feature][Doc] Add more source and sink examples of sort-standalone
vernedeng opened a new issue, #562: URL: https://github.com/apache/inlong-website/issues/562 ### Description _No response_ ### Use case _No response_ ### Are you willing to submit PR? - [X] Yes, I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] GanfengTan commented on a diff in pull request #5780: [INLONG-5722][Agent] Support Redis Source
GanfengTan commented on code in PR #5780:
URL: https://github.com/apache/inlong/pull/5780#discussion_r972564411
##
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/RedisReader.java:
##
@@ -0,0 +1,445 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.agent.plugin.sources.reader;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.TypeAdapter;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonWriter;
+import com.moilioncircle.redis.replicator.RedisReplicator;
+import com.moilioncircle.redis.replicator.Replicator;
+import com.moilioncircle.redis.replicator.cmd.CommandName;
+import com.moilioncircle.redis.replicator.cmd.impl.DefaultCommand;
+import com.moilioncircle.redis.replicator.cmd.parser.DefaultCommandParser;
+import com.moilioncircle.redis.replicator.event.Event;
+import com.moilioncircle.redis.replicator.event.EventListener;
+import com.moilioncircle.redis.replicator.event.PostRdbSyncEvent;
+import com.moilioncircle.redis.replicator.rdb.datatype.KeyStringValueHash;
+import com.moilioncircle.redis.replicator.rdb.datatype.KeyStringValueList;
+import com.moilioncircle.redis.replicator.rdb.datatype.KeyStringValueSet;
+import com.moilioncircle.redis.replicator.rdb.datatype.KeyStringValueString;
+import com.moilioncircle.redis.replicator.rdb.datatype.KeyStringValueZSet;
+import com.moilioncircle.redis.replicator.rdb.datatype.KeyValuePair;
+import com.moilioncircle.redis.replicator.rdb.datatype.ZSetEntry;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.message.DefaultMessage;
+import org.apache.inlong.agent.metrics.audit.AuditUtils;
+import org.apache.inlong.agent.plugin.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Redis data reader
+ */
+public class RedisReader extends AbstractReader {
+
+public static final String REDIS_READER_TAG_NAME = "AgentRedisMetric";
+public static final String JOB_REDIS_PORT = "job.redisJob.port";
+public static final String JOB_REDIS_HOSTNAME = "job.redisJob.hostname";
+public static final String JOB_REDIS_SSL = "job.redisJob.ssl";
+public static final String JOB_REDIS_AUTHUSER = "job.redisJob.authUser";
+public static final String JOB_REDIS_AUTHPASSWORD =
"job.redisJob.authPassword";
+public static final String JOB_REDIS_READTIMEOUT =
"job.redisJob.readTimeout";
+public static final String JOB_REDIS_QUEUE_SIZE = "job.redisJob.queueSize";
+public static final String JOB_REDIS_REPLID = "job.redisJob.replId";
+public static final String JOB_REDIS_OFFSET = "job.redisJob.offset";
+private static final Logger LOGGER =
LoggerFactory.getLogger(RedisReader.class);
+private String port;
+private String hostName;
+private boolean ssl;
+private String authUser;
+private String authPassword;
+private String readTimeout;
+private String instanceId;
+private String replId;
+private String snapShot;
+private boolean destroyed;
+private Replicator redisReplicator;
+private LinkedBlockingQueue redisMessageQueue;
+private boolean finished = false;
+private ExecutorService executor;
+private Gson gson;
+
Review Comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5802: [INLONG-5495][SDK] Support multi-topic manager
vernedeng commented on code in PR #5802:
URL: https://github.com/apache/inlong/pull/5802#discussion_r972565660
##
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongMultiTopicManager.java:
##
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sdk.sort.manager;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InlongTopicTypeEnum;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.TopicFetcher;
+import org.apache.inlong.sdk.sort.api.TopicFetcherBuilder;
+import org.apache.inlong.sdk.sort.api.TopicManager;
+import org.apache.inlong.sdk.sort.entity.ConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.fetcher.tube.TubeConsumerCreator;
+import org.apache.inlong.sdk.sort.util.PeriodicTask;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.inlong.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.exception.TubeClientException;
+import org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
+import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * Inlong manager that maintain the {@link
org.apache.inlong.sdk.sort.api.MultiTopicsFetcher}.
+ * It is suitable to the cases that topics share the same configurations.
+ * And each consumer will consume multi topic.
+ */
+public class InlongMultiTopicManager extends TopicManager {
+
+private static final Logger LOGGER =
LoggerFactory.getLogger(InlongMultiTopicManager.class);
+
+private final Map> pulsarFetchers = new
ConcurrentHashMap<>();
+private final Map> kafkaFetchers = new
ConcurrentHashMap<>();
+private final Map> tubeFetchers = new
ConcurrentHashMap<>();
+private final Map allFetchers = new
ConcurrentHashMap<>();
+private Set allTopics = new HashSet<>();
+private final PeriodicTask updateMetaDataWorker;
+
+private boolean stopAssign = false;
+private int consumerSize;
+
+public InlongMultiTopicManager(ClientContext context, QueryConsumeConfig
queryConsumeConfig) {
+super(context, queryConsumeConfig);
+this.consumerSize = context.getConfig().getMaxConsumerSize();
+updateMetaDataWorker = new
UpdateMetaDataThread(context.getConfig().getUpdateMetaDataIntervalSec(),
+TimeUnit.SECONDS);
+String threadName = "sortsdk_multi_topic_manager_" +
context.getConfig().getSortTaskId()
++ "_" + StringUtil.formatDate(new Date(), "-MM-dd
HH:mm:ss");
+updateMetaDataWorker.start(threadName);
+LOGGER.info("create InlongMultiTopicManager success");
+}
+
+@Override
+public boolean clean() {
+LOGGER.info("start clean {}", context.getConfig().getSortTaskId());
+close();
+offlineAllTopicsAndPartitions();
+LOGGER.info("end clean {}", context.getConfig().getSortTaskId());
+return true;
+}
+
+@Override
+public TopicFetcher addTopic(InLongTopic topic) {
+return null;
+}
+
+@Override
+public TopicFetcher removeTopic(InLongTopic topic, boolean closeFetcher) {
+return null;
+}
+
+@Override
+public TopicFetcher getFetcher(String fetchKey) {
+return allFetchers.get(fetchKey);
+}
+
+@Override
+public Collection getAllFetchers() {
+return allFetchers.values();
+}
+
+@Override
+public Set getManagedInLongTopics() {
+return allTopics;
+}
+
+@Override
+public void offlineAl
[GitHub] [inlong-website] dockerzhang commented on a diff in pull request #559: [INLONG-551][Doc] Update release doc and fix bugs
dockerzhang commented on code in PR #559:
URL: https://github.com/apache/inlong-website/pull/559#discussion_r972571927
##
i18n/zh-CN/docusaurus-plugin-content-docs-community/current/how-to-release.md:
##
@@ -405,7 +405,8 @@ svn delete
https://dist.apache.org/repos/dist/release/inlong/${last_release_vers
### 更新官网链接
### 发ANNOUNCE邮件，主送 `d...@inlong.apache.org`，抄送 `annou...@apache.org`
-**请确保6.4中的仓库已发布成功，一般是在6.4后的24小时后发布邮件**
+**请确保6.4中的仓库已发布成功，一般是在6.4后的24小时后发布邮件**
+**确保已经发布成功的方式为登陆 https://downloads.apache.org/inlong/${release-version}/
查看是否有安装包**
Review Comment:
**确保已经发布成功的方式为登陆 https://downloads.apache.org/inlong/${release-version}/
查看是否有安装包**
->
登陆 https://downloads.apache.org/inlong/${release-version}/ 查看是否有安装包**
##
community/how-to-release.md:
##
@@ -31,7 +31,8 @@ $ gpg --version #check the version, should be 2.x
### Generate gpg Key
Attentions：
- Name is best to keep consistent with your full name of Apache ID
-- Email should be the Apache email
+- The mailbox used must be an apache mailbox, it is recommended to check all
keys with gpg -k first
+- Make sure that there is only one key, and delete the keys of other spare
mailboxes first
Review Comment:
no this restriction?
##
i18n/zh-CN/docusaurus-plugin-content-docs-community/current/how-to-release.md:
##
@@ -29,7 +29,7 @@ $ gpg --version #检查版本，应该为2.x
### 生成 gpg Key
需要注意以下几点：
- 输入名字时最好与 Apache 中登记的 Full name 保持一致
-- 使用的邮箱应该是 apache 邮箱
+- 使用的邮箱必须是 apache 邮箱，建议先 gpg -k 查看所有 key, 确保只有一个 key，多余的其他邮箱的 key 先删除
Review Comment:
'，多余的其他邮箱的 key 先删除' no this restriction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] gong commented on a diff in pull request #5879: [INLONG-5133][Sort] Support influxDB sink for sort
gong commented on code in PR #5879: URL: https://github.com/apache/inlong/pull/5879#discussion_r972572495 ## inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/InfluxDBLoadNode.java: ## @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.inlong.sort.protocol.node.load; + +import javax.annotation.Nullable; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.NoArgsConstructor; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName; +import org.apache.inlong.sort.protocol.FieldInfo; +import org.apache.inlong.sort.protocol.InlongMetric; +import org.apache.inlong.sort.protocol.enums.FilterStrategy; +import org.apache.inlong.sort.protocol.node.LoadNode; +import org.apache.inlong.sort.protocol.transformation.FieldRelation; +import org.apache.inlong.sort.protocol.transformation.FilterFunction; + +import java.io.Serializable; +import java.util.List; +import java.util.Map; +import org.apache.inlong.sort.protocol.transformation.WatermarkField; + +/** + * MySql load node can load data into MySql + */ Review Comment: description is error -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #5906: [INLONG-5903][Sort] Make InLong metric constructs factory more cohesive
yunqingmoswu commented on code in PR #5906:
URL: https://github.com/apache/inlong/pull/5906#discussion_r972569971
##
inlong-sort/sort-connectors/base/src/main/java/org/apache/inlong/sort/base/metric/SinkMetricData.java:
##
@@ -55,21 +56,40 @@ public class SinkMetricData implements MetricData {
private Meter numRecordsOutPerSecond;
private Meter numBytesOutPerSecond;
-public SinkMetricData(String groupId, String streamId, String nodeId,
MetricGroup metricGroup) {
-this(groupId, streamId, nodeId, metricGroup, null);
-}
-
public SinkMetricData(MetricOption option, MetricGroup metricGroup) {
-this(option.getGroupId(), option.getStreamId(), option.getNodeId(),
metricGroup, option.getIpPorts());
+this(option.getGroupId(), option.getStreamId(), option.getNodeId(),
+option.getRegisteredMetric(), metricGroup,
option.getIpPorts());
}
public SinkMetricData(
-String groupId, String streamId, String nodeId, MetricGroup
metricGroup,
+@Nullable String groupId,
+@Nullable String streamId,
+@Nullable String nodeId,
+@Nullable RegisteredMetric registeredMetric,
+MetricGroup metricGroup,
@Nullable String auditHostAndPorts) {
this.metricGroup = metricGroup;
-this.groupId = groupId;
-this.streamId = streamId;
-this.nodeId = nodeId;
+if (groupId != null && streamId != null && nodeId != null) {
+this.groupId = groupId;
+this.streamId = streamId;
+this.nodeId = nodeId;
+if (RegisteredMetric.ALL.equals(registeredMetric)) {
Review Comment:
RegisteredMetric.ALL.equals(registeredMetric) -> RegisteredMetric.ALL ==
registeredMetric
##
inlong-sort/sort-connectors/base/src/main/java/org/apache/inlong/sort/base/metric/MetricOption.java:
##
@@ -86,4 +91,49 @@ public HashSet getIpPortList() {
public String getIpPorts() {
return ipPorts;
}
+
+public RegisteredMetric getRegisteredMetric() {
+return registeredMetric;
+}
+
+public static Builder builder() {
+return new Builder();
+}
+
+public enum RegisteredMetric {
+ALL,
+NORMAL,
+DIRTY
+}
+
+public static class Builder {
+private String inLongMetric;
+private String inLongAudit;
Review Comment:
inLongAudit -> inlongAudit
##
inlong-sort/sort-connectors/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java:
##
@@ -41,44 +43,44 @@
*/
public class SourceMetricData implements MetricData {
-private final MetricGroup metricGroup;
-private final String groupId;
-private final String streamId;
-private final String nodeId;
+private MetricGroup metricGroup;
+private String groupId;
+private String streamId;
+private String nodeId;
private Counter numRecordsIn;
private Counter numBytesIn;
private Meter numRecordsInPerSecond;
private Meter numBytesInPerSecond;
-private final AuditImp auditImp;
-
-public SourceMetricData(String groupId, String streamId, String nodeId,
MetricGroup metricGroup) {
-this(groupId, streamId, nodeId, metricGroup, (AuditImp) null);
-}
+private AuditImp auditImp;
public SourceMetricData(MetricOption option, MetricGroup metricGroup) {
-this(option.getGroupId(), option.getStreamId(), option.getNodeId(),
metricGroup, option.getIpPorts());
+this(option.getGroupId(), option.getStreamId(), option.getNodeId(),
+option.getRegisteredMetric(), metricGroup,
option.getIpPorts());
}
-public SourceMetricData(String groupId, String streamId, String nodeId,
MetricGroup metricGroup,
-AuditImp auditImp) {
-this.groupId = groupId;
-this.streamId = streamId;
-this.nodeId = nodeId;
-this.metricGroup = metricGroup;
-this.auditImp = auditImp;
-}
-
-public SourceMetricData(String groupId, String streamId, String nodeId,
MetricGroup metricGroup,
+public SourceMetricData(
+@Nullable String groupId,
+@Nullable String streamId,
+@Nullable String nodeId,
+@Nullable RegisteredMetric registeredMetric,
+MetricGroup metricGroup,
@Nullable String auditHostAndPorts) {
-this.groupId = groupId;
-this.streamId = streamId;
-this.nodeId = nodeId;
this.metricGroup = metricGroup;
+if (groupId != null && streamId != null && nodeId != null) {
+this.groupId = groupId;
+this.streamId = streamId;
+this.nodeId = nodeId;
+if (RegisteredMetric.ALL.equals(registeredMetric)) {
Review Comment:
RegisteredMetric.ALL.equals(registeredMetric) -> RegisteredMetric.ALL ==
registeredMetric
##
inlong-sort/sort-connectors/base/src/m
[GitHub] [inlong] dockerzhang commented on a diff in pull request #5912: [INLONG-5911][Dashboard] Modify the data group name to the data stream group
dockerzhang commented on code in PR #5912: URL: https://github.com/apache/inlong/pull/5912#discussion_r972579420 ## inlong-dashboard/src/locales/cn.json: ## @@ -196,16 +196,16 @@ "meta.Group.MessageMiddleware": "消息中间件", "meta.Group.AccessSize": "按天接入大小", "meta.Group.GB/Day": "GB/天", - "meta.Group.BusinessIntroduction": "分组描述", - "meta.Group.BusinessOwners": "分组责任人", + "meta.Group.BusinessIntroduction": "流组描述", + "meta.Group.BusinessOwners": "流组责任人", Review Comment: Group 责任人 ## inlong-dashboard/src/locales/cn.json: ## @@ -196,16 +196,16 @@ "meta.Group.MessageMiddleware": "消息中间件", "meta.Group.AccessSize": "按天接入大小", "meta.Group.GB/Day": "GB/天", - "meta.Group.BusinessIntroduction": "分组描述", - "meta.Group.BusinessOwners": "分组责任人", + "meta.Group.BusinessIntroduction": "流组描述", + "meta.Group.BusinessOwners": "流组责任人", "meta.Group.SingleStripMaximumLength": "单条最大长度", "meta.Group.thousand/day": "万条/天", "meta.Group.AccessPeakPerSecond": "每秒接入峰值", - "meta.Group.InlongGroupId": "分组ID", + "meta.Group.InlongGroupId": "流组ID", "meta.Group.InlongGroupIdRules": "只能包含小写字母、数字、中划线、下划线", - "meta.Group.InlongGroupName": "分组名称", + "meta.Group.InlongGroupName": "流组名称", Review Comment: Group 名称 ## inlong-dashboard/src/locales/cn.json: ## @@ -312,7 +312,7 @@ "pages.GroupDetail.Info.SubmittedSuccessfully": "提交成功", "pages.GroupDetail.Info.CheckMsg": "请检查表单", "pages.GroupDetail.Info.Previous": "上一步", - "pages.GroupDetail.Info.Create": "新建分组", + "pages.GroupDetail.Info.Create": "新建流组", Review Comment: ditto ## inlong-dashboard/src/locales/cn.json: ## @@ -196,16 +196,16 @@ "meta.Group.MessageMiddleware": "消息中间件", "meta.Group.AccessSize": "按天接入大小", "meta.Group.GB/Day": "GB/天", - "meta.Group.BusinessIntroduction": "分组描述", - "meta.Group.BusinessOwners": "分组责任人", + "meta.Group.BusinessIntroduction": "流组描述", + "meta.Group.BusinessOwners": "流组责任人", "meta.Group.SingleStripMaximumLength": "单条最大长度", "meta.Group.thousand/day": "万条/天", "meta.Group.AccessPeakPerSecond": "每秒接入峰值", - "meta.Group.InlongGroupId": "分组ID", + "meta.Group.InlongGroupId": "流组ID", "meta.Group.InlongGroupIdRules": "只能包含小写字母、数字、中划线、下划线", - "meta.Group.InlongGroupName": "分组名称", + "meta.Group.InlongGroupName": "流组名称", "meta.Group.NumberOfAccess": "按天接入条数", - "meta.Group.BusinessOwnersExtra": "分组责任人，可查看、修改分组信息", + "meta.Group.BusinessOwnersExtra": "流组责任人，可查看、修改流组信息", Review Comment: Group 责任人，可查看、修改 Group 信息 ## inlong-dashboard/src/locales/cn.json: ## @@ -196,16 +196,16 @@ "meta.Group.MessageMiddleware": "消息中间件", "meta.Group.AccessSize": "按天接入大小", "meta.Group.GB/Day": "GB/天", - "meta.Group.BusinessIntroduction": "分组描述", - "meta.Group.BusinessOwners": "分组责任人", + "meta.Group.BusinessIntroduction": "流组描述", Review Comment: Group 描述 ## inlong-dashboard/src/locales/cn.json: ## @@ -196,16 +196,16 @@ "meta.Group.MessageMiddleware": "消息中间件", "meta.Group.AccessSize": "按天接入大小", "meta.Group.GB/Day": "GB/天", - "meta.Group.BusinessIntroduction": "分组描述", - "meta.Group.BusinessOwners": "分组责任人", + "meta.Group.BusinessIntroduction": "流组描述", + "meta.Group.BusinessOwners": "流组责任人", "meta.Group.SingleStripMaximumLength": "单条最大长度", "meta.Group.thousand/day": "万条/天", "meta.Group.AccessPeakPerSecond": "每秒接入峰值", - "meta.Group.InlongGroupId": "分组ID", + "meta.Group.InlongGroupId": "流组ID", Review Comment: Group ID ## inlong-dashboard/src/locales/cn.json: ## @@ -303,7 +303,7 @@ "pages.GroupDashboard.config.Reject": "已驳回", "pages.GroupDashboard.config.ExecuteLog": "执行日志", "pages.GroupDashboard.ConfirmDelete": "确认删除吗", - "pages.GroupDashboard.Create": "新建分组", + "pages.GroupDashboard.Create": "新建流组", Review Comment: 新建 Group ## inlong-dashboard/src/locales/cn.json: ## @@ -258,7 +258,7 @@ "meta.Consumption.ConsumerGroupNameRules": "只能包含小写字母、数字、中划线、下划线", "meta.Consumption.DataStreamIDsHelp": "多个数据流ID之间用逗号(,)隔开", "meta.Consumption.ConsumerDataStreamID": "消费的数据流ID", - "meta.Consumption.ConsumerTargetBusinessID": "消费的数据分组ID", + "meta.Consumption.ConsumerTargetBusinessID": "消费的数据流组ID", Review Comment: 消费的数据流 Group ID ## inlong-dashboard/src/locales/cn.json: ## @@ -395,7 +395,7 @@ "pages.Approvals.Approver": "审批人", "pages.Approvals.Applicant": "申请人", "pages.Approvals.ProcessID": "流程单ID", - "pages.Approvals.GroupId": "数据分组ID", + "pages.Approvals.GroupId": "数据流组ID", Review Comment: 数据流 Group ID ## inlong-dashboard/src/locales/cn.json: ## @@ -411,7 +411,7 @@ "pages.ConsumeDashboard.config.Total": "消费总数", "pages.ConsumeDashboard.config.Pending": "待审批", "pages.ConsumeDashboard.config.ConsumerGroup": "消费组", - "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的分组ID", + "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #5879: [INLONG-5133][Sort] Support influxDB sink for sort
yunqingmoswu commented on code in PR #5879: URL: https://github.com/apache/inlong/pull/5879#discussion_r972577506 ## inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java: ## @@ -72,7 +73,8 @@ @JsonSubTypes.Type(value = OracleLoadNode.class, name = "oracleLoad"), @JsonSubTypes.Type(value = GreenplumLoadNode.class, name = "greenplumLoad"), @JsonSubTypes.Type(value = DLCIcebergLoadNode.class, name = "dlcIcebergLoad"), -@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad") +@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad"), +@JsonSubTypes.Type(value = InfluxDBLoadNode.class, name = "influxdb") Review Comment: influxdb -> influxdbLoad -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on pull request #5864: [INLONG-5863][Manager] Extend Redis extract node
yunqingmoswu commented on PR #5864: URL: https://github.com/apache/inlong/pull/5864#issuecomment-1248880798 > If you want to use redis extract node in sort, you must implement the protocol of sort with redis, see `org.apache.inlong.manager.pojo.sort.util.ExtractNodeUtils` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on pull request #5864: [INLONG-5863][Manager] Extend Redis extract node
yunqingmoswu commented on PR #5864: URL: https://github.com/apache/inlong/pull/5864#issuecomment-1248881910 > > Maybe miss the part that interacts with the sort protocol？ > > I follow the [extend ExtractNode](https://inlong.apache.org/zh-CN/docs/design_and_concept/how_to_extend_data_node_for_manager) guideline. It already had the redis extract node. What I do is add manager plugin thus I add the following classes:RedisSource, RedisSourceDTO, RedisRequestDTO, RedisSourceOperator, add Redis type inTaskTypeEnum and SourceType. If you want to use it in sort you must implement the protocol of sort with redis, see `org.apache.inlong.manager.pojo.sort.util.ExtractNodeUtils` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] bluewang commented on pull request #387: [INLONG-1831][Website] Add official website content search button
bluewang commented on PR #387: URL: https://github.com/apache/inlong-website/pull/387#issuecomment-1248886800 > @dockerzhang @bluewang @leezng I found the search engine seems do not to work. Let me check it. Is there any progress on this situation? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #5774: [INLONG-5133][Sort] Support Apache InfluxDB source
yunqingmoswu commented on code in PR #5774:
URL: https://github.com/apache/inlong/pull/5774#discussion_r972586581
##
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java:
##
@@ -62,7 +63,8 @@
@JsonSubTypes.Type(value = TubeMQExtractNode.class, name =
"tubeMQExtract"),
@JsonSubTypes.Type(value = PulsarExtractNode.class, name =
"pulsarExtract"),
@JsonSubTypes.Type(value = RedisExtractNode.class, name =
"redisExtract"),
-@JsonSubTypes.Type(value = DorisExtractNode.class, name =
"dorisExtract")
+@JsonSubTypes.Type(value = DorisExtractNode.class, name =
"dorisExtract"),
+@JsonSubTypes.Type(value = InfluxDBExtractNode.class, name =
"influxDBExtract")
Review Comment:
Please keep the naming consistent of influxExtract between the Node and
ExtractNode.
##
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java:
##
@@ -75,6 +76,7 @@
@JsonSubTypes.Type(value = TubeMQExtractNode.class, name =
"tubeMQExtract"),
@JsonSubTypes.Type(value = RedisExtractNode.class, name =
"redisExtract"),
@JsonSubTypes.Type(value = DorisExtractNode.class, name =
"dorisExtract"),
+@JsonSubTypes.Type(value = InfluxDBExtractNode.class, name =
"influxExtract"),
Review Comment:
Please keep the naming consistent of `influxExtract` between the Node and
ExtractNode.
##
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/InfluxDBExtractNode.java:
##
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.protocol.node.extract;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.inlong.common.enums.MetaField;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.InlongMetric;
+import org.apache.inlong.sort.protocol.Metadata;
+import org.apache.inlong.sort.protocol.node.ExtractNode;
+import org.apache.inlong.sort.protocol.transformation.WatermarkField;
+
+import javax.annotation.Nullable;
+import java.io.Serializable;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Extract node for influxDB, note that InfluxDB should work in replicaSet mode
+ */
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeName("influxDBExtract")
Review Comment:
Please keep the naming consistent of influxExtract between the Node and
ExtractNode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5912: [INLONG-5911][Dashboard] Modify the data group name to the data stream group
healchow commented on code in PR #5912: URL: https://github.com/apache/inlong/pull/5912#discussion_r972588612 ## inlong-dashboard/src/locales/cn.json: ## @@ -411,7 +411,7 @@ "pages.ConsumeDashboard.config.Total": "消费总数", "pages.ConsumeDashboard.config.Pending": "待审批", "pages.ConsumeDashboard.config.ConsumerGroup": "消费组", - "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的分组ID", + "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的流组ID", Review Comment: 感觉会和上面的“消费组”存在歧义。 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang commented on a diff in pull request #5912: [INLONG-5911][Dashboard] Modify the data group name to the data stream group
dockerzhang commented on code in PR #5912: URL: https://github.com/apache/inlong/pull/5912#discussion_r972588846 ## inlong-dashboard/src/locales/cn.json: ## @@ -411,7 +411,7 @@ "pages.ConsumeDashboard.config.Total": "消费总数", "pages.ConsumeDashboard.config.Pending": "待审批", "pages.ConsumeDashboard.config.ConsumerGroup": "消费组", - "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的分组ID", + "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的流组ID", Review Comment: 消费的 Group ID -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] liangyepianzhou commented on a diff in pull request #5879: [INLONG-5133][Sort] Support influxDB sink for sort
liangyepianzhou commented on code in PR #5879:
URL: https://github.com/apache/inlong/pull/5879#discussion_r972589473
##
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java:
##
@@ -72,7 +73,8 @@
@JsonSubTypes.Type(value = OracleLoadNode.class, name = "oracleLoad"),
@JsonSubTypes.Type(value = GreenplumLoadNode.class, name =
"greenplumLoad"),
@JsonSubTypes.Type(value = DLCIcebergLoadNode.class, name =
"dlcIcebergLoad"),
-@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad")
+@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad"),
+@JsonSubTypes.Type(value = InfluxDBLoadNode.class, name = "influxdb")
Review Comment:
1. Name changes done.
2.` the package of jdbc/src/main... should not be under sort-connectors.`,
do you mean `the package of influxdb-jdbc/src/main`. And please explain why
this is wrong.
3. this CDC sues JDBC table API: `options.put("connector", "jdbc-inlong");`
4. @jun0315 is already testing the referenced influx-jdbc-driver connector
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #5879: [INLONG-5133][Sort] Support influxDB sink for sort
yunqingmoswu commented on code in PR #5879: URL: https://github.com/apache/inlong/pull/5879#discussion_r972593293 ## inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java: ## @@ -72,7 +73,8 @@ @JsonSubTypes.Type(value = OracleLoadNode.class, name = "oracleLoad"), @JsonSubTypes.Type(value = GreenplumLoadNode.class, name = "greenplumLoad"), @JsonSubTypes.Type(value = DLCIcebergLoadNode.class, name = "dlcIcebergLoad"), -@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad") +@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad"), +@JsonSubTypes.Type(value = InfluxDBLoadNode.class, name = "influxdb") Review Comment: 2.If it is based on the jdbc connector extension, you can put the influxdb-jdbc code in the jdbc connector, there is no need to start a single module 3.If it already have the unit test, you can put the corresponding link. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang commented on a diff in pull request #5912: [INLONG-5911][Dashboard] Modify the data group name to the data stream group
dockerzhang commented on code in PR #5912: URL: https://github.com/apache/inlong/pull/5912#discussion_r972603846 ## inlong-dashboard/src/locales/cn.json: ## @@ -411,7 +411,7 @@ "pages.ConsumeDashboard.config.Total": "消费总数", "pages.ConsumeDashboard.config.Pending": "待审批", "pages.ConsumeDashboard.config.ConsumerGroup": "消费组", - "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的分组ID", + "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的流组ID", Review Comment: ```suggestion "pages.ConsumeDashboard.config.ConsumptionInlongGroupId": "消费的Group ID", ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] gosonzhang opened a new pull request, #5916: [INLONG-5915][DataProxy] Optimize source's ServerMessageHandler class
gosonzhang opened a new pull request, #5916: URL: https://github.com/apache/inlong/pull/5916 - Fixes #5915 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #5912: [INLONG-5911][Dashboard] Modify the data group name to the data stream group
dockerzhang merged PR #5912: URL: https://github.com/apache/inlong/pull/5912 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated: [INLONG-5911][Dashboard] Modify the data group name to the data stream group (#5912)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 3e6a1c80f [INLONG-5911][Dashboard] Modify the data group name to the
data stream group (#5912)
3e6a1c80f is described below
commit 3e6a1c80fb597e9659861cab3e299eb699cd5815
Author: Daniel
AuthorDate: Fri Sep 16 13:11:54 2022 +0800
[INLONG-5911][Dashboard] Modify the data group name to the data stream
group (#5912)
Co-authored-by: Charles Zhang
---
inlong-dashboard/src/i18n.ts | 8 +++-
inlong-dashboard/src/locales/cn.json | 22 +++---
2 files changed, 14 insertions(+), 16 deletions(-)
diff --git a/inlong-dashboard/src/i18n.ts b/inlong-dashboard/src/i18n.ts
index af4b6dd30..d77cd3ce1 100644
--- a/inlong-dashboard/src/i18n.ts
+++ b/inlong-dashboard/src/i18n.ts
@@ -26,9 +26,8 @@ const resources = {
en: {
translation: {
'configs.menus.Process': 'Approval',
- 'configs.menus.Groups': 'Data Group',
- 'configs.menus.DataConsumption': 'Data Consumption',
- 'configs.menus.Datasources': 'Data Sources',
+ 'configs.menus.Groups': 'Data Groups',
+ 'configs.menus.DataConsumption': 'Data Consumptions',
'configs.menus.Clusters': 'Clusters',
'configs.menus.ClusterTags': 'ClusterTags',
'configs.menus.SystemManagement': 'System',
@@ -40,9 +39,8 @@ const resources = {
cn: {
translation: {
'configs.menus.Process': '审批管理',
- 'configs.menus.Groups': '数据分组',
+ 'configs.menus.Groups': '数据流组',
'configs.menus.DataConsumption': '数据消费',
- 'configs.menus.Datasources': '数据源服务器',
'configs.menus.Clusters': '集群管理',
'configs.menus.ClusterTags': '标签管理',
'configs.menus.SystemManagement': '系统管理',
diff --git a/inlong-dashboard/src/locales/cn.json
b/inlong-dashboard/src/locales/cn.json
index 4c95c3055..d1d8dcd22 100644
--- a/inlong-dashboard/src/locales/cn.json
+++ b/inlong-dashboard/src/locales/cn.json
@@ -196,16 +196,16 @@
"meta.Group.MessageMiddleware": "消息中间件",
"meta.Group.AccessSize": "按天接入大小",
"meta.Group.GB/Day": "GB/天",
- "meta.Group.BusinessIntroduction": "分组描述",
- "meta.Group.BusinessOwners": "分组责任人",
+ "meta.Group.BusinessIntroduction": "描述",
+ "meta.Group.BusinessOwners": "责任人",
"meta.Group.SingleStripMaximumLength": "单条最大长度",
"meta.Group.thousand/day": "万条/天",
"meta.Group.AccessPeakPerSecond": "每秒接入峰值",
- "meta.Group.InlongGroupId": "分组ID",
+ "meta.Group.InlongGroupId": "Group ID",
"meta.Group.InlongGroupIdRules": "只能包含小写字母、数字、中划线、下划线",
- "meta.Group.InlongGroupName": "分组名称",
+ "meta.Group.InlongGroupName": "数据流组名称",
"meta.Group.NumberOfAccess": "按天接入条数",
- "meta.Group.BusinessOwnersExtra": "分组责任人，可查看、修改分组信息",
+ "meta.Group.BusinessOwnersExtra": "责任人，可查看、修改数据流组信息",
"meta.Group.QueueModule": "队列模型",
"meta.Group.Parallel": "并行",
"meta.Group.Serial": "顺序",
@@ -258,7 +258,7 @@
"meta.Consumption.ConsumerGroupNameRules": "只能包含小写字母、数字、中划线、下划线",
"meta.Consumption.DataStreamIDsHelp": "多个数据流ID之间用逗号(,)隔开",
"meta.Consumption.ConsumerDataStreamID": "消费的数据流ID",
- "meta.Consumption.ConsumerTargetBusinessID": "消费的数据分组ID",
+ "meta.Consumption.ConsumerTargetBusinessID": "消费的 Group ID",
"meta.Consumption.No": "否",
"meta.Consumption.filterEnabled": "是否过滤消费",
"meta.Consumption.MasterAddress": "Master地址",
@@ -303,7 +303,7 @@
"pages.GroupDashboard.config.Reject": "已驳回",
"pages.GroupDashboard.config.ExecuteLog": "执行日志",
"pages.GroupDashboard.ConfirmDelete": "确认删除吗",
- "pages.GroupDashboard.Create": "新建分组",
+ "pages.GroupDashboard.Create": "新建数据流组",
"pages.GroupDashboard.SuccessfullyDeleted": "删除成功",
"pages.GroupDetail.Info.Back": "返回",
"pages.GroupDetail.Info.NextStep": "下一步",
@@ -312,7 +312,7 @@
"pages.GroupDetail.Info.SubmittedSuccessfully": "提交成功",
"pages.GroupDetail.Info.CheckMsg": "请检查表单",
"pages.GroupDetail.Info.Previous": "上一步",
- "pages.GroupDetail.Info.Create": "新建分组",
+ "pages.GroupDetail.Info.Create": "新建数据流组",
"pages.GroupDetail.Sources.SaveSuccessfully": "保存成功",
"pages.GroupDetail.Sources.DeleteConfirm": "确认删除吗",
"pages.GroupDetail.Sources.DeleteSuccessfully": "删除成功",
@@ -343,7 +343,7 @@
"pages.GroupDetail.Stream.StreamConfigTitle": "数据流配置",
"pages.GroupDetail.Stream.CreateDataStream": "新建数据流",
"pages.GroupDetail.PageTitle": "详情",
- "pages.GroupDetail.Info": "数据分组",
+ "pages.GroupDetail.Info": "数据流组",
"pages.GroupDetail.Streams": "数据流",
"pages.GroupDetail.Sources": "数据源",
"pages.GroupDetail.Sinks": "数据存储",
@@ -395,7 +395,7 @@
"pages.Approvals.Approver": "审批人",
"pages.Approvals.Applicant": "申请人",
"pages.Approvals.ProcessID": "流程单ID",
- "pages.Approvals.GroupId": "数据分组ID",
+ "pages.Approvals.GroupId": "Group ID",
"pages.Approvals.MyApplication": "我的申请",
"pages.Approvals.MyApproval": "我的审批",
[GitHub] [inlong-website] Chever-John commented on pull request #387: [INLONG-1831][Website] Add official website content search button
Chever-John commented on PR #387: URL: https://github.com/apache/inlong-website/pull/387#issuecomment-1248950575 @bluewang Give me a few days, I'll give it a try. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] liangyepianzhou commented on a diff in pull request #5879: [INLONG-5133][Sort] Support influxDB sink for sort
liangyepianzhou commented on code in PR #5879: URL: https://github.com/apache/inlong/pull/5879#discussion_r972641320 ## inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java: ## @@ -72,7 +73,8 @@ @JsonSubTypes.Type(value = OracleLoadNode.class, name = "oracleLoad"), @JsonSubTypes.Type(value = GreenplumLoadNode.class, name = "greenplumLoad"), @JsonSubTypes.Type(value = DLCIcebergLoadNode.class, name = "dlcIcebergLoad"), -@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad") +@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad"), +@JsonSubTypes.Type(value = InfluxDBLoadNode.class, name = "influxdb") Review Comment: > 2.If it is based on the jdbc connector extension, you can put the influxdb-jdbc code in the jdbc connector, there is no need to start a single module I have a problem with this, this is a jdbc-driver. rather than a jdbc table API. The JDBC-connector module calls the influx-jdbc-driver module. When the influx jdbc driver is officially supported, we should delete this module and replace it with a new jdbc-driver. Writing them together not only makes no sense but also increases code coupling. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #5879: [INLONG-5133][Sort] Support influxDB sink for sort
yunqingmoswu commented on code in PR #5879: URL: https://github.com/apache/inlong/pull/5879#discussion_r972651568 ## inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java: ## @@ -72,7 +73,8 @@ @JsonSubTypes.Type(value = OracleLoadNode.class, name = "oracleLoad"), @JsonSubTypes.Type(value = GreenplumLoadNode.class, name = "greenplumLoad"), @JsonSubTypes.Type(value = DLCIcebergLoadNode.class, name = "dlcIcebergLoad"), -@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad") +@JsonSubTypes.Type(value = DorisLoadNode.class, name = "dorisLoad"), +@JsonSubTypes.Type(value = InfluxDBLoadNode.class, name = "influxdb") Review Comment: The Sort Connectors module contains all connectors or modules common to connectors. There is currently no official driver implementation for influx db. It is reasonable to put the corresponding driver implementation in the jdbc connector. When the official implementation is available later, remove it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] bluewang commented on pull request #387: [INLONG-1831][Website] Add official website content search button
bluewang commented on PR #387: URL: https://github.com/apache/inlong-website/pull/387#issuecomment-1248966203 > @bluewang Give me a few days, I'll give it a try. @Chever-John OK. Waiting for your good news -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #5785: [INLONG-5784][Sort] Add metric state for PostgreSQL
EMsnap commented on code in PR #5785:
URL: https://github.com/apache/inlong/pull/5785#discussion_r972656995
##
inlong-sort/sort-connectors/base/src/main/java/org/apache/inlong/sort/base/util/MetricStateUtils.java:
##
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.base.util;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.inlong.sort.base.metric.MetricState;
+import org.apache.inlong.sort.base.metric.SourceMetricData;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN;
+import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_IN;
+
+/**
+ * metric state for {@link MetricState} supporting snapshot and restore
+ */
+@Slf4j
+public class MetricStateUtils {
+
+/**
+ *
+ * restore metric state data
+ * @param metricStateListState state data list
+ * @param subtaskIndex current subtask index
+ * @param currentSubtaskNum number of current parallel subtask
+ * @return metric state
+ * @throws Exception throw exception metricStateListState.get()
+ */
+public static MetricState restoreMetricState(ListState
metricStateListState, Integer subtaskIndex,
+Integer currentSubtaskNum) throws Exception {
+if (metricStateListState == null || metricStateListState.get() ==
null) {
+return null;
+}
+log.info("restoreMetricState:{}, subtaskIndex:{},
currentSubtaskNum:{}", metricStateListState, subtaskIndex,
+currentSubtaskNum);
+MetricState currentMetricState;
+Map map = new HashMap<>(16);
+for (MetricState metricState : metricStateListState.get()) {
+map.put(metricState.getSubtaskIndex(), metricState);
+}
+int previousSubtaskNum = map.size();
+if (currentSubtaskNum >= previousSubtaskNum) {
+currentMetricState = map.get(subtaskIndex);
+} else {
+Map metrics = new HashMap<>(4);
+currentMetricState = new MetricState(subtaskIndex, metrics);
+List indexList = computeIndexList(subtaskIndex,
currentSubtaskNum, previousSubtaskNum);
+for (Integer index : indexList) {
+MetricState metricState = map.get(index);
+for (Map.Entry entry :
metricState.getMetrics().entrySet()) {
+if (metrics.containsKey(entry.getKey())) {
+metrics.put(entry.getKey(),
metrics.get(entry.getKey()) + entry.getValue());
+} else {
+metrics.put(entry.getKey(), entry.getValue());
+}
+}
+}
+}
+return currentMetricState;
+}
+
+/**
+ *
+ * Assignment previous subtask index to current subtask when reduce
parallelism
+ * @param subtaskIndex current subtask index
+ * @param currentSubtaskNum number of current parallel subtask
+ * @param previousSubtaskNum number of previous parallel subtask
+ * @return index list
+ */
+public static List computeIndexList(Integer subtaskIndex, Integer
currentSubtaskNum,
Review Comment:
the doc should explain the rule on how are the indexs are assigned
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] iamsee123 commented on pull request #5864: [INLONG-5863][Manager] Extend Redis extract node
iamsee123 commented on PR #5864: URL: https://github.com/apache/inlong/pull/5864#issuecomment-1248968819 > > > Maybe miss the part that interacts with the sort protocol？ > > > > > > I follow the [extend ExtractNode](https://inlong.apache.org/zh-CN/docs/design_and_concept/how_to_extend_data_node_for_manager) guideline. It already had the redis extract node. What I do is add manager plugin thus I add the following classes:RedisSource, RedisSourceDTO, RedisRequestDTO, RedisSourceOperator, add Redis type inTaskTypeEnum and SourceType. > > If you want to use it in sort you must implement the protocol of sort with redis, see `org.apache.inlong.manager.pojo.sort.util.ExtractNodeUtils` Fixed. Thank you for your comment. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] luchunliang commented on a diff in pull request #5843: [INLONG-5842][Manager] Support maintenance of message queue cluster
luchunliang commented on code in PR #5843:
URL: https://github.com/apache/inlong/pull/5843#discussion_r972667887
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/queue/MessageQueueServiceImpl.java:
##
@@ -0,0 +1,555 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.service.cluster.queue;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.inlong.common.msg.AttributeConstants;
+import org.apache.inlong.manager.common.consts.MQType;
+import org.apache.inlong.manager.dao.entity.InlongClusterEntity;
+import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
+import org.apache.inlong.manager.dao.mapper.InlongClusterEntityMapper;
+import org.apache.inlong.manager.dao.mapper.InlongGroupEntityMapper;
+import
org.apache.inlong.manager.pojo.cluster.queue.MessageQueueClearTopicRequest;
+import org.apache.inlong.manager.pojo.cluster.queue.MessageQueueControlRequest;
+import org.apache.inlong.manager.pojo.cluster.queue.MessageQueueOfflineRequest;
+import org.apache.inlong.manager.pojo.cluster.queue.MessageQueueOnlineRequest;
+import
org.apache.inlong.manager.pojo.cluster.queue.MessageQueueSynchronizeTopicRequest;
+import org.apache.inlong.manager.pojo.cluster.tubemq.TubeClusterInfo;
+import org.apache.inlong.manager.pojo.group.InlongGroupPageRequest;
+import org.apache.inlong.manager.service.resource.queue.tubemq.TubeMQOperator;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.DeleteTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.Topics;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * Inlong message queue cluster operator.
+ */
+@Service
+public class MessageQueueServiceImpl implements MessageQueueService {
+
+public static final Logger LOG =
LoggerFactory.getLogger(MessageQueueServiceImpl.class);
+
+public static final Splitter.MapSplitter MAP_SPLITTER =
Splitter.on(AttributeConstants.SEPARATOR)
+
.trimResults().withKeyValueSeparator(AttributeConstants.KEY_VALUE_SEPARATOR);
+public static final Joiner.MapJoiner MAP_JOINER =
Joiner.on(AttributeConstants.SEPARATOR)
+.withKeyValueSeparator(AttributeConstants.KEY_VALUE_SEPARATOR);
+public static final String KEY_PRODUCER = "producer";
+public static final String KEY_CONSUMER = "consumer";
+public static final String KEY_ADMIN_URL = "adminUrl";
+public static final String KEY_AUTHENTICATION = "authentication";
+public static final String KEY_TENANT = "tenant";
+public static final String KEY_NAMESPACE = "namespace";
+public static final String KEY_NUM_PARTITIONS = "numPartitions";
+public static final String KEY_REPLICATION_FACTOR = "replicationFactor";
+public static final int DEFAULT_NUM_PARTITIONS = 10;
+public static final short DEFAULT_REPLICATION_FACTOR = 2;
+
+@Autowired
+private InlongClusterEntityMapper clusterMapper;
+@Autowired
+private InlongGroupEntityMapper groupMapper;
+@Autowired
+private TubeMQOperator tubeMQOperator;
+
+/**
+ * Control produce operation and consume operation of Inlong message queue
cluster
+ */
+@Override
+public String control(MessageQueueControlRequest request) {
+String name = request.getName();
+// check parameters
+if (StringUtils.isEmpty(name)) {
+return "miss cluster na
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #5879: [INLONG-5133][Sort] Support influxDB sink for sort
yunqingmoswu commented on code in PR #5879:
URL: https://github.com/apache/inlong/pull/5879#discussion_r972673256
##
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/InfluxDBLoadNode.java:
##
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.protocol.node.load;
+
+import javax.annotation.Nullable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.InlongMetric;
+import org.apache.inlong.sort.protocol.enums.FilterStrategy;
+import org.apache.inlong.sort.protocol.node.LoadNode;
+import org.apache.inlong.sort.protocol.transformation.FieldRelation;
+import org.apache.inlong.sort.protocol.transformation.FilterFunction;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import org.apache.inlong.sort.protocol.transformation.WatermarkField;
+
+/**
+ * influxDB load node can load data into InfluxDB.
+ */
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeName("influxDBLoadNode")
Review Comment:
influxDBLoadNode -> influxdbLoad
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
