Re: [PR] [INLONG-11129][Sort] Enhanced Metric Instrumentation for InLong Sort Flink Connector [inlong]
PeterZh6 commented on PR #11130: URL: https://github.com/apache/inlong/pull/11130#issuecomment-2364981190 > > @PeterZh6 please fix the failed workflow, thanks. > > Thanks for following up with this. However, I am unable to reproduce the test failure locally, so may still need some time. The failure is linked to [[Bug][Sort] Mongodb2StarRocksTest Failure Due to Potential Dependency Conflicts #11166](https://github.com/apache/inlong/issues/11166) I am currently working on a solution and will submit a pull request to address the problem as soon as possible. -- 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
Re: [PR] [INLONG-10463][SDK] Optimization of ultra-long field processing in InlongSDK [inlong]
qy-liuhuo commented on PR #9: URL: https://github.com/apache/inlong/pull/9#issuecomment-2364975137 > > > > > How to successfully parse incomplete data? > > > > > > > > > > > > Now when the configuration does not allow truncation, if the body exceeds the limit, it will be rejected directly. > > > > Specifically for a body of type `byte[]`, `SendResult.BODY_EXCEED_MAX_LEN` will be returned directly. And for `List` type bodies, they are truncated at the body granularity, which ensures the integrity of each body. > > > > > > > > > In any case, the SDK should not set truncation, otherwise this operation will modify the data content reported by the business and cause errors > > > > > > If it cannot be truncated, is this issue unnecessary? > > It cannot be truncated, this is the premise. How to check whether it is too long requires analysis to see how to modify it appropriately. I have fixed it, now if the data is too long, it will directly return `BODY_EXCEED_MAX_LEN` 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
Re: [PR] [INLONG-11148][Sort] Fixes Log4j2 of sort-end-to-end-test-v1.15 not working properly [inlong]
PeterZh6 commented on PR #11149: URL: https://github.com/apache/inlong/pull/11149#issuecomment-2364981257 > > @PeterZh6 please fix the failed workflows, thanks. > > Thanks for following up with this. However, I am unable to reproduce the test failure locally, so may still need some time The failure is linked to [[Bug][Sort] Mongodb2StarRocksTest Failure Due to Potential Dependency Conflicts #11166](https://github.com/apache/inlong/issues/11166) I am currently working on a solution and will submit a pull request to address the problem as soon as possible. -- 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
Re: [PR] [INLONG-10464][SDK] InlongSDK support retry sending when failed [inlong]
gosonzhang commented on code in PR #11144:
URL: https://github.com/apache/inlong/pull/11144#discussion_r1768260102
##
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java:
##
@@ -191,11 +195,93 @@ public String getSDKVersion() {
return ConfigConstants.PROXY_SDK_VERSION;
}
+public void setSendSuccess(boolean sendSuccess) {
+this.sendSuccess = sendSuccess;
+}
+
+public boolean isSendSuccess() {
+return sendSuccess;
+}
+
+private SendResult retryWhenSendMessageFail(Function sendOperation,
+DefaultMessageSender initialSender) {
+int attempts = 0;
+SendResult sendResult = null;
+DefaultMessageSender currentSender = initialSender;
+while (attempts < ConfigConstants.SENDER_MAX_RETRY) {
+sendResult = sendOperation.apply(currentSender);
+if (sendResult != null && sendResult.equals(SendResult.OK)) {
+currentSender.setSendSuccess(true);
+return sendResult;
+}
+currentSender.setSendSuccess(false);
+// try to get success sender
+DefaultMessageSender randomSuccessSender =
getRandomSuccessSender();
+if (randomSuccessSender != null) {
+currentSender = randomSuccessSender;
+} else {
+break;
+}
+attempts++;
+}
+
+return sendResult;
+}
+
+private String
retryWhenSendMessageIndexFail(Function
sendOperation,
+DefaultMessageSender initialSender) {
+int attempts = 0;
+String sendIndexResult = null;
+DefaultMessageSender currentSender = initialSender;
+while (attempts < ConfigConstants.SENDER_MAX_RETRY) {
+sendIndexResult = sendOperation.apply(currentSender);
+if (sendIndexResult != null &&
sendIndexResult.startsWith(SendResult.OK.toString())) {
+currentSender.setSendSuccess(true);
+return sendIndexResult;
+}
+currentSender.setSendSuccess(false);
+// try to get success sender
+DefaultMessageSender randomSuccessSender =
getRandomSuccessSender();
+if (randomSuccessSender != null) {
+currentSender = randomSuccessSender;
+} else {
+break;
+}
+attempts++;
+}
+
+return sendIndexResult;
+}
+
+private DefaultMessageSender getRandomSuccessSender() {
+List keys = new ArrayList<>(CACHE_SENDER.keySet());
+if (keys.isEmpty()) {
+return null;
+}
+int attempts = 0;
+int maxAttempts = keys.size();
+// choose sending success MessageSender randomly
+while (attempts < maxAttempts) {
+int randomIndex = ThreadLocalRandom.current().nextInt(keys.size());
Review Comment:
The senders are saved according to different clusterIds and cannot be
obtained randomly.
--
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
Re: [PR] [INLONG-10464][SDK] InlongSDK support retry sending when failed [inlong]
gosonzhang commented on code in PR #11144:
URL: https://github.com/apache/inlong/pull/11144#discussion_r1768263223
##
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/ConfigConstants.java:
##
@@ -74,4 +74,6 @@ public class ConfigConstants {
public static String HTTP = "http://";;
public static String HTTPS = "https://";;
+public static int SENDER_MAX_RETRY = 3;
Review Comment:
The number of retries should be configurable
--
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
Re: [PR] [INLONG-11065][Sort] Provides a method to add openTelemetryAppender for the sort connector [inlong]
vernedeng commented on code in PR #11066:
URL: https://github.com/apache/inlong/pull/11066#discussion_r1768118911
##
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/util/OpenTelemetryLogger.java:
##
@@ -0,0 +1,145 @@
+/*
+ * 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 io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporter;
+import
io.opentelemetry.instrumentation.log4j.appender.v2_17.OpenTelemetryAppender;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.logs.SdkLoggerProvider;
+import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.semconv.resource.attributes.ResourceAttributes;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.Layout;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.core.layout.PatternLayout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+
+public class OpenTelemetryLogger {
+
+private OpenTelemetrySdk SDK; // OpenTelemetry SDK
+
+private final String endpoint; // OpenTelemetry Exporter Endpoint
+
+private final String serviceName; // OpenTelemetry Service Name
+
+private final Layout layout; // Log4j Layout
+
+private final Level logLevel; // Log4j Log Level
+
+private static final Logger LOG =
LoggerFactory.getLogger(OpenTelemetryLogger.class);
+
+public OpenTelemetryLogger() {
+// Default Service Name
+serviceName = "inlong-sort-connector";
+// Get OpenTelemetry Exporter Endpoint from Environment Variable
+if (System.getenv("OTEL_EXPORTER_ENDPOINT") != null) {
+endpoint = System.getenv("OTEL_EXPORTER_ENDPOINT");
+} else {
+endpoint = "localhost:4317";
+}
+// Default Log4j Layout
+this.layout = PatternLayout.newBuilder()
+.withPattern("%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} -
%msg%n")
+.withCharset(StandardCharsets.UTF_8)
+.build();
+// Default Log4j Log Level
+this.logLevel = Level.INFO;
+}
+
+public OpenTelemetryLogger(String serviceName, String endpoint, Layout
layout, Level logLevel) {
+this.serviceName = serviceName;
+this.endpoint = endpoint;
+this.layout = layout;
+this.logLevel = logLevel;
+}
+
+private void createOpenTelemetrySdk() {
Review Comment:
optimize the format 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
Re: [PR] [INLONG-11123][SDK] Transform SQL supports conv function [inlong]
dockerzhang merged PR #11141: URL: https://github.com/apache/inlong/pull/11141 -- 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-11150][Manager] Fix the problem of incorrect setting of sorTaskName for sink (#11151)
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 0d0061 [INLONG-11150][Manager] Fix the problem of incorrect
setting of sorTaskName for sink (#11151)
0d0061 is described below
commit 0d0061b29b36b4152dd066494aae67473e4b
Author: fuweng11 <76141879+fuwen...@users.noreply.github.com>
AuthorDate: Fri Sep 20 17:25:30 2024 +0800
[INLONG-11150][Manager] Fix the problem of incorrect setting of sorTaskName
for sink (#11151)
---
.../resource/sink/AbstractStandaloneSinkResourceOperator.java | 10 +-
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/AbstractStandaloneSinkResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/AbstractStandaloneSinkResourceOperator.java
index d5a97a42b1..d1842fb02c 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/AbstractStandaloneSinkResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/AbstractStandaloneSinkResourceOperator.java
@@ -89,13 +89,13 @@ public abstract class
AbstractStandaloneSinkResourceOperator implements SinkReso
if (!StringUtils.isAnyBlank(sinkInfo.getSortConsumerGroup(),
sinkInfo.getSortTaskName())) {
return;
}
-if (StringUtils.isBlank(sinkInfo.getSortConsumerGroup())) {
-sinkInfo.setSortConsumerGroup(sinkInfo.getDataNodeName());
+StreamSinkEntity sink =
sinkEntityMapper.selectByPrimaryKey(sinkInfo.getId());
+if (StringUtils.isBlank(sink.getSortConsumerGroup())) {
+sink.setSortConsumerGroup(sink.getDataNodeName());
}
-if (StringUtils.isBlank(sinkInfo.getSortTaskName())) {
-sinkInfo.setSortTaskName(sinkInfo.getDataNodeName());
+if (StringUtils.isBlank(sink.getSortTaskName())) {
+sink.setSortTaskName(sink.getDataNodeName());
}
-StreamSinkEntity sink =
sinkEntityMapper.selectByPrimaryKey(sinkInfo.getId());
sinkEntityMapper.updateByIdSelective(sink);
}
Re: [PR] [INLONG-11129][Sort] Enhanced Metric Instrumentation for InLong Sort Flink Connector [inlong]
dockerzhang commented on PR #11130: URL: https://github.com/apache/inlong/pull/11130#issuecomment-2363276706 @PeterZh6 please fix the failed workflow, thanks. -- 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
Re: [PR] [INLONG-11148][Sort] Fixes Log4j2 of sort-end-to-end-test-v1.15 not working properly [inlong]
dockerzhang commented on PR #11149: URL: https://github.com/apache/inlong/pull/11149#issuecomment-2363274317 @PeterZh6 please fix the failed workflows, thanks. -- 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-11123][SDK] Transform SQL supports conv function (#11141)
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 86635a83ba [INLONG-11123][SDK] Transform SQL supports conv function
(#11141)
86635a83ba is described below
commit 86635a83ba11cb9996640d277363a37e2ccd10f1
Author: Zkplo <87751516+zk...@users.noreply.github.com>
AuthorDate: Fri Sep 20 17:26:55 2024 +0800
[INLONG-11123][SDK] Transform SQL supports conv function (#11141)
Co-authored-by: ZKpLo <14148880+zk...@user.noreply.gitee.com>
---
.../process/function/RadixConvertFunction.java | 97 ++
.../arithmetic/testRadixConvertFunction.java | 87 +++
2 files changed, 184 insertions(+)
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/RadixConvertFunction.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/RadixConvertFunction.java
new file mode 100644
index 00..f9b3cac523
--- /dev/null
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/RadixConvertFunction.java
@@ -0,0 +1,97 @@
+/*
+ * 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.transform.process.function;
+
+import org.apache.inlong.sdk.transform.decode.SourceData;
+import org.apache.inlong.sdk.transform.process.Context;
+import org.apache.inlong.sdk.transform.process.operator.OperatorTools;
+import org.apache.inlong.sdk.transform.process.parser.ValueParser;
+
+import lombok.extern.slf4j.Slf4j;
+import net.sf.jsqlparser.expression.Expression;
+import net.sf.jsqlparser.expression.Function;
+
+import java.math.BigInteger;
+import java.util.List;
+
+/**
+ * RadixConvertFunction -> radix_convert(N,from_base,to_base)
+ * description: Converts numbers between different number bases. The minimum
base is 2 and the maximum base is 36.
+ * If from_base is a negative number, N is regarded as a signed
number. Otherwise, N is treated as
+ * unsigned. This function works with 64-bit precision.
+ * - returns NULL if any of its arguments are NULL.
+ * - returns conversion results otherwise.
+ */
+@Slf4j
+@TransformFunction(names = {"radix_convert"})
+public class RadixConvertFunction implements ValueParser {
+
+private final ValueParser numParser;
+private final ValueParser fromBaseParser;
+private final ValueParser toBaseParser;
+
+public RadixConvertFunction(Function expr) {
+List expressions = expr.getParameters().getExpressions();
+numParser = OperatorTools.buildParser(expressions.get(0));
+fromBaseParser = OperatorTools.buildParser(expressions.get(1));
+toBaseParser = OperatorTools.buildParser(expressions.get(2));
+}
+
+@Override
+public Object parse(SourceData sourceData, int rowIndex, Context context) {
+Object numObj = numParser.parse(sourceData, rowIndex, context);
+Object fromBaseObj = fromBaseParser.parse(sourceData, rowIndex,
context);
+Object toBaseObj = toBaseParser.parse(sourceData, rowIndex, context);
+if (numObj == null || fromBaseObj == null || toBaseObj == null) {
+return null;
+}
+String num = OperatorTools.parseString(numObj);
+int fromBase = Integer.parseInt(fromBaseObj.toString());
+int toBase = Integer.parseInt(toBaseObj.toString());
+return conv(num, fromBase, toBase);
+}
+
+/**
+ * Converts a number from one base to another.
+ *
+ * @param number The number to be converted (as a string or integer).
+ * @param fromBase The base of the input number.
+ * @param toBase The base to which the number should be converted.
+ * @return The converted number as a string, or null if input is invalid.
+ */
+public static String conv(String number, int fromBase, int toBase) {
+if (!checkRange(fromBase) || !checkRange(toBase)) {
+return null;
+}
+try {
+BigInteger num = new BigInteger(number, Math.abs(fromBase));
+
Re: [PR] [INLONG-11150][Manager] Fix the problem of incorrect setting of sorTaskName for sink [inlong]
dockerzhang merged PR #11151: URL: https://github.com/apache/inlong/pull/11151 -- 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
Re: [PR] [INLONG-11107][SDK] Avro Source Data Support Map Type [inlong]
dockerzhang merged PR #11108: URL: https://github.com/apache/inlong/pull/11108 -- 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-11107][SDK] Avro Source Data Support Map Type (#11108)
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 ca4069293c [INLONG-11107][SDK] Avro Source Data Support Map Type
(#11108)
ca4069293c is described below
commit ca4069293c9110a90f80f6262504f1c4639c2815
Author: Xincheng Huang <60057611+ying-...@users.noreply.github.com>
AuthorDate: Fri Sep 20 17:28:12 2024 +0800
[INLONG-11107][SDK] Avro Source Data Support Map Type (#11108)
---
.../inlong/sdk/transform/decode/AvroSourceData.java | 12
.../process/processor/AbstractProcessorTestBase.java | 19 +++
.../process/processor/TestAvro2CsvProcessor.java | 9 +
3 files changed, 28 insertions(+), 12 deletions(-)
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/AvroSourceData.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/AvroSourceData.java
index c060c89af4..42705433f4 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/AvroSourceData.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/AvroSourceData.java
@@ -20,12 +20,14 @@ package org.apache.inlong.sdk.transform.decode;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Type;
import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.util.Utf8;
import org.apache.commons.lang3.StringUtils;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
public class AvroSourceData implements SourceData {
@@ -91,6 +93,16 @@ public class AvroSourceData implements SourceData {
// parse other node
for (int i = 1; i < childNodes.size(); i++) {
AvroNode node = childNodes.get(i);
+if (curSchema.getType() == Type.MAP) {
+Map map = (Map) current;
+Object mapValue = map.get(new Utf8(node.getName()));
+if (mapValue == null) {
+return "";
+}
+curSchema = curSchema.getValueType();
+current = mapValue;
+continue;
+}
if (curSchema.getType() != Type.RECORD) {
// error data
return "";
diff --git
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/AbstractProcessorTestBase.java
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/AbstractProcessorTestBase.java
index 9f05396667..0a71004789 100644
---
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/AbstractProcessorTestBase.java
+++
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/AbstractProcessorTestBase.java
@@ -67,14 +67,17 @@ public abstract class AbstractProcessorTestBase {
}
protected byte[] getAvroTestData() {
-String srcString =
"T2JqAQIWYXZyby5zY2hlbWHIBXsidHlwZSI6InJlY29yZCIsIm5hbWUiOiJTZGtEYXRhUmVxdWVzdCIs"
-+
"Im5hbWVzcGFjZSI6InRlc3QiLCJmaWVsZHMiOlt7Im5hbWUiOiJzaWQiLCJ0eXBlIjoic3RyaW5nIn0seyJuYW1lIj"
-+
"oibXNncyIsInR5cGUiOnsidHlwZSI6ImFycmF5IiwiaXRlbXMiOnsidHlwZSI6InJlY29yZCIsIm5hbWUiOiJTZGtNZ"
-+
"XNzYWdlIiwiZmllbGRzIjpbeyJuYW1lIjoibXNnIiwidHlwZSI6ImJ5dGVzIn0seyJuYW1lIjoibXNnVGltZSIsInR5"
-+
"cGUiOiJsb25nIn0seyJuYW1lIjoiZXh0aW5mbyIsInR5cGUiOnsidHlwZSI6Im1hcCIsInZhbHVlcyI6InN0cmluZyJ"
-+
"9fV19fX0seyJuYW1lIjoicGFja2FnZUlEIiwidHlwZSI6ImxvbmcifV19AI7h/J8SaFCGp012msD3lKMCngEIc2lkMQ"
-+
"QKQXBwbGXyhcYJBAhrZXkxCGtleTEIa2V5Mgx2YWx1ZTEADEJhbmFuYeSLjBMECGtleTEIa2V5MghrZXkyDHZhbHVlM"
-+ "gAAgIkPjuH8nxJoUIanTXaawPeUow==";
+String srcString =
"T2JqAQIWYXZyby5zY2hlbWHIBXsidHlwZSI6InJlY29yZCIsIm5hbWUiOiJTZGtE"
++
"YXRhUmVxdWVzdCIsIm5hbWVzcGFjZSI6InRlc3QiLCJmaWVsZHMiOlt7Im5hbWUi"
++
"OiJzaWQiLCJ0eXBlIjoic3RyaW5nIn0seyJuYW1lIjoibXNncyIsInR5cGUiOnsi"
++
"dHlwZSI6ImFycmF5IiwiaXRlbXMiOnsidHlwZSI6InJlY29yZCIsIm5hbWUiOiJT"
++
"ZGtNZXNzYWdlIiwiZmllbGRzIjpbeyJuYW1lIjoibXNnIiwidHlwZSI6ImJ5dGVz"
++
"In0seyJuYW1lIjoibXNnVGltZSIsInR5cGUiOiJsb25nIn0seyJuYW1lIjoiZXh0"
++
"aW5mbyIsInR5cGUiOnsidHlwZSI6Im1hcCIsInZhbHVlcyI6InN0cmluZyJ9fV19"
++
"fX0seyJuYW1lIjoicGFja2FnZUlEIiwidHlwZSI6ImxvbmcifV19AMt7kQjpgkXl"
++
"EjM4Iv+oOJYClgEIc2lkMQQKQXBwbGXyhcYJBARrMQx2YWx1ZTEEazIMdmFsdWUy"
++
"AAxCYW5hbmHki4wTBARrMQx2YWx1ZTMEazIMdmFsdWU0AACAiQ/Le5EI6YJF5RIz"
++ "OCL/qDiW";
[PR] [INLONG-11163][Manager] Fix the problem of adding dataaddtask failed [inlong]
fuweng11 opened a new pull request, #11164: URL: https://github.com/apache/inlong/pull/11164 Fixes #11163 ### Motivation Fix the problem of adding dataaddtask failed. ### Modifications Fix the problem of adding dataaddtask failed. -- 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
Re: [PR] [INLONG-10463][SDK] Optimization of ultra-long field processing in InlongSDK [inlong]
gosonzhang commented on PR #9: URL: https://github.com/apache/inlong/pull/9#issuecomment-2363146077 > > > > How to successfully parse incomplete data? > > > > > > > > > Now when the configuration does not allow truncation, if the body exceeds the limit, it will be rejected directly. > > > Specifically for a body of type `byte[]`, `SendResult.BODY_EXCEED_MAX_LEN` will be returned directly. And for `List` type bodies, they are truncated at the body granularity, which ensures the integrity of each body. > > > > > > In any case, the SDK should not set truncation, otherwise this operation will modify the data content reported by the business and cause errors > > If it cannot be truncated, is this issue unnecessary? It cannot be truncated, this is the premise. How to check whether it is too long requires analysis to see how to modify it appropriately. -- 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-11060][SDK] Transform support REGEXP_...() related functions (#11102)
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 cc06854292 [INLONG-11060][SDK] Transform support REGEXP_...() related
functions (#11102)
cc06854292 is described below
commit cc06854292e6fc1318135711f669f363018d9f91
Author: emptyOVO <118812562+empty...@users.noreply.github.com>
AuthorDate: Fri Sep 20 17:29:30 2024 +0800
[INLONG-11060][SDK] Transform support REGEXP_...() related functions
(#11102)
---
.../process/function/RegexpCountFunction.java | 71
.../process/function/RegexpExtractAllFunction.java | 94 ++
.../process/function/RegexpExtractFunction.java| 84 +++
.../transform/process/function/RegexpFunction.java | 65 +++
.../process/function/RegexpInstrFunction.java | 71
.../process/function/RegexpReplaceFunction.java| 67 +++
.../process/function/RegexpSubstrFunction.java | 70
.../function/string/TestRegexExtractFunction.java | 91 +
.../function/string/TestRegexpCountFunction.java | 77 ++
.../string/TestRegexpExtractAllFunction.java | 92 +
.../function/string/TestRegexpFunction.java| 77 ++
.../function/string/TestRegexpInstrFunction.java | 84 +++
.../function/string/TestRegexpReplaceFunction.java | 65 +++
.../function/string/TestRegexpSubstrFunction.java | 84 +++
14 files changed, 1092 insertions(+)
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/RegexpCountFunction.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/RegexpCountFunction.java
new file mode 100644
index 00..bc7091a0ce
--- /dev/null
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/RegexpCountFunction.java
@@ -0,0 +1,71 @@
+/*
+ * 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.transform.process.function;
+
+import org.apache.inlong.sdk.transform.decode.SourceData;
+import org.apache.inlong.sdk.transform.process.Context;
+import org.apache.inlong.sdk.transform.process.operator.OperatorTools;
+import org.apache.inlong.sdk.transform.process.parser.ValueParser;
+
+import net.sf.jsqlparser.expression.Expression;
+import net.sf.jsqlparser.expression.Function;
+
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * RegexpCountFunction
+ * description: REGEXP_COUNT(str, regexp)--Returns the number of times str
matches the regexp pattern.
+ * regexp must be a Java regular expression.
+ * Returns an INTEGER representation of the number of matches.
+ * NULL if any of the arguments are NULL or regexp is invalid.
+ */
+@TransformFunction(names = {"regexp_count"})
+public class RegexpCountFunction implements ValueParser {
+
+private ValueParser inputStringParser;
+
+private ValueParser patternStringParser;
+
+public RegexpCountFunction(Function expr) {
+if (expr.getParameters() != null) {
+List expressions =
expr.getParameters().getExpressions();
+if (expressions != null && expressions.size() >= 2) {
+inputStringParser =
OperatorTools.buildParser(expressions.get(0));
+patternStringParser =
OperatorTools.buildParser(expressions.get(1));
+}
+}
+}
+
+@Override
+public Object parse(SourceData sourceData, int rowIndex, Context context) {
+if (inputStringParser == null || patternStringParser == null) {
+return null;
+}
+String inputString =
OperatorTools.parseString(inputStringParser.parse(sourceData, rowIndex,
context));
+String patternString =
OperatorTools.parseString(patternStringParser.parse(sourceData, rowIndex,
context));
+Pattern pattern = Pattern.compile(patternString);
+Matcher ma
Re: [PR] [INLONG-11060][SDK] Transform support REGEXP_...() related functions [inlong]
dockerzhang merged PR #11102: URL: https://github.com/apache/inlong/pull/11102 -- 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-11005][SDK] Add YAML formatted data source for Transform (#11085)
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 260a12d8dd [INLONG-11005][SDK] Add YAML formatted data source for
Transform (#11085)
260a12d8dd is described below
commit 260a12d8dd80196551c2783809b62cf5d4218b10
Author: emptyOVO <118812562+empty...@users.noreply.github.com>
AuthorDate: Fri Sep 20 17:30:08 2024 +0800
[INLONG-11005][SDK] Add YAML formatted data source for Transform (#11085)
---
.../sdk/transform/decode/SourceDecoderFactory.java | 6 +
.../inlong/sdk/transform/decode/YamlNode.java | 35 ++
.../sdk/transform/decode/YamlSourceData.java | 99
.../sdk/transform/decode/YamlSourceDecoder.java| 121 +++
.../inlong/sdk/transform/pojo/YamlSourceInfo.java | 53 +
.../process/processor/TestYaml2CsvProcessor.java | 130 +
6 files changed, 444 insertions(+)
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/SourceDecoderFactory.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/SourceDecoderFactory.java
index 2fcab7c0bc..041e1632eb 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/SourceDecoderFactory.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/SourceDecoderFactory.java
@@ -23,6 +23,7 @@ import org.apache.inlong.sdk.transform.pojo.CsvSourceInfo;
import org.apache.inlong.sdk.transform.pojo.JsonSourceInfo;
import org.apache.inlong.sdk.transform.pojo.KvSourceInfo;
import org.apache.inlong.sdk.transform.pojo.PbSourceInfo;
+import org.apache.inlong.sdk.transform.pojo.YamlSourceInfo;
public class SourceDecoderFactory {
@@ -48,4 +49,9 @@ public class SourceDecoderFactory {
public static BsonSourceDecoder createBsonDecoder(BsonSourceInfo
sourceInfo) {
return new BsonSourceDecoder(sourceInfo);
}
+
+public static YamlSourceDecoder createYamlDecoder(YamlSourceInfo
sourceInfo) {
+return new YamlSourceDecoder(sourceInfo);
+}
+
}
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlNode.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlNode.java
new file mode 100644
index 00..097ef854d1
--- /dev/null
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlNode.java
@@ -0,0 +1,35 @@
+/*
+ * 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.transform.decode;
+
+import lombok.Data;
+
+@Data
+public class YamlNode {
+
+private String name;
+private Object value;
+
+public YamlNode() {
+}
+
+public YamlNode(String name, Object value) {
+this.name = name;
+this.value = value;
+}
+}
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlSourceData.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlSourceData.java
new file mode 100644
index 00..a9d3162fd7
--- /dev/null
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlSourceData.java
@@ -0,0 +1,99 @@
+/*
+ * 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.a
Re: [PR] [INLONG-11005][SDK] Add YAML formatted data source for Transform [inlong]
dockerzhang merged PR #11085: URL: https://github.com/apache/inlong/pull/11085 -- 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
Re: [PR] [INLONG-10464][SDK] InlongSDK support retry sending when failed [inlong]
emptyOVO commented on code in PR #11144:
URL: https://github.com/apache/inlong/pull/11144#discussion_r1768423100
##
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java:
##
@@ -191,11 +195,93 @@ public String getSDKVersion() {
return ConfigConstants.PROXY_SDK_VERSION;
}
+public void setSendSuccess(boolean sendSuccess) {
+this.sendSuccess = sendSuccess;
+}
+
+public boolean isSendSuccess() {
+return sendSuccess;
+}
+
+private SendResult retryWhenSendMessageFail(Function sendOperation,
+DefaultMessageSender initialSender) {
+int attempts = 0;
+SendResult sendResult = null;
+DefaultMessageSender currentSender = initialSender;
+while (attempts < ConfigConstants.SENDER_MAX_RETRY) {
+sendResult = sendOperation.apply(currentSender);
+if (sendResult != null && sendResult.equals(SendResult.OK)) {
+currentSender.setSendSuccess(true);
+return sendResult;
+}
+currentSender.setSendSuccess(false);
+// try to get success sender
+DefaultMessageSender randomSuccessSender =
getRandomSuccessSender();
+if (randomSuccessSender != null) {
+currentSender = randomSuccessSender;
+} else {
+break;
+}
+attempts++;
+}
+
+return sendResult;
+}
+
+private String
retryWhenSendMessageIndexFail(Function
sendOperation,
+DefaultMessageSender initialSender) {
+int attempts = 0;
+String sendIndexResult = null;
+DefaultMessageSender currentSender = initialSender;
+while (attempts < ConfigConstants.SENDER_MAX_RETRY) {
+sendIndexResult = sendOperation.apply(currentSender);
+if (sendIndexResult != null &&
sendIndexResult.startsWith(SendResult.OK.toString())) {
+currentSender.setSendSuccess(true);
+return sendIndexResult;
+}
+currentSender.setSendSuccess(false);
+// try to get success sender
+DefaultMessageSender randomSuccessSender =
getRandomSuccessSender();
+if (randomSuccessSender != null) {
+currentSender = randomSuccessSender;
+} else {
+break;
+}
+attempts++;
+}
+
+return sendIndexResult;
+}
+
+private DefaultMessageSender getRandomSuccessSender() {
+List keys = new ArrayList<>(CACHE_SENDER.keySet());
+if (keys.isEmpty()) {
+return null;
+}
+int attempts = 0;
+int maxAttempts = keys.size();
+// choose sending success MessageSender randomly
+while (attempts < maxAttempts) {
+int randomIndex = ThreadLocalRandom.current().nextInt(keys.size());
Review Comment:
sorry i made a mistake, now i change the logic to choosing another
send-success sender in `CACHE_SENDER` directly
--
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
[PR] [INLONG-10119][SDK] Supporting Data Sharding with GroupBy Semantics [inlong]
emptyOVO opened a new pull request, #11165: URL: https://github.com/apache/inlong/pull/11165 Fixes #10119 ### Motivation * add `groupByItems` to parse `group by` Semantics * Because there is no aggregate function related implementation, so therefore, only simple grouping logic is currently implemented * provide unit tests with examples of json data ### Modifications ### Verifying this change *(Please pick either of the following options)* - [ ] This change is a trivial rework/code cleanup without any test coverage. - [x] This change is already covered by existing tests, such as: *(please describe tests)* - [ ] 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 follow-up 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
Re: [PR] [INLONG-11156][SDK] SortSDK support that the token configuration of pulsar cluster is null [inlong]
luchunliang merged PR #11158: URL: https://github.com/apache/inlong/pull/11158 -- 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-11156][SDK] SortSDK support that the token configuration of pulsar cluster is null (#11158)
This is an automated email from the ASF dual-hosted git repository.
luchunliang 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 f1ccde411e [INLONG-11156][SDK] SortSDK support that the token
configuration of pulsar cluster is null (#11158)
f1ccde411e is described below
commit f1ccde411e4693830c9ac7d7799f1941fbe560b0
Author: ChunLiang Lu
AuthorDate: Fri Sep 20 15:35:57 2024 +0800
[INLONG-11156][SDK] SortSDK support that the token configuration of pulsar
cluster is null (#11158)
---
.../inlong/sdk/sort/manager/InlongMultiTopicManager.java | 12 +++-
.../sdk/sort/manager/InlongSingleTopicManager.java | 16 +++-
.../inlong/sdk/sort/manager/InlongTopicManager.java | 12 +++-
3 files changed, 33 insertions(+), 7 deletions(-)
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongMultiTopicManager.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongMultiTopicManager.java
index 744b38d223..2d098345ca 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongMultiTopicManager.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongMultiTopicManager.java
@@ -35,6 +35,7 @@ import
org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
@@ -213,10 +214,19 @@ public class InlongMultiTopicManager extends TopicManager
{
topic.getInLongCluster().getBootstraps(), consumerSize);
for (int i = 0; i < consumerSize; i++) {
try {
+String token = topic.getInLongCluster().getToken();
+Authentication auth = null;
+if (StringUtils.isNoneBlank(token)) {
+auth = AuthenticationFactory.token(token);
+}
PulsarClient pulsarClient = PulsarClient.builder()
.serviceUrl(topic.getInLongCluster().getBootstraps())
-
.authentication(AuthenticationFactory.token(topic.getInLongCluster().getToken()))
+.authentication(auth)
.build();
+LOGGER.info("create pulsar client succ cluster:{}, topic:{},
token:{}",
+topic.getInLongCluster().getClusterId(),
+topic.getTopic(),
+topic.getInLongCluster().getToken());
TopicFetcher fetcher = TopicFetcherBuilder.newPulsarBuilder()
.pulsarClient(pulsarClient)
.topic(topics)
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java
index 503304c77f..d137a903d2 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java
@@ -33,6 +33,7 @@ import
org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClient;
import org.slf4j.Logger;
@@ -363,15 +364,20 @@ public class InlongSingleTopicManager extends
TopicManager {
if
(!pulsarClients.containsKey(topic.getInLongCluster().getClusterId())) {
if (topic.getInLongCluster().getBootstraps() != null) {
try {
+String token = topic.getInLongCluster().getToken();
+Authentication auth = null;
+if (StringUtils.isNoneBlank(token)) {
+auth = AuthenticationFactory.token(token);
+}
PulsarClient pulsarClient = PulsarClient.builder()
.serviceUrl(topic.getInLongCluster().getBootstraps())
-
.authentication(AuthenticationFactory.token(topic.getInLongCluster().getToken()))
+.authentication(auth)
.build();
pulsarClients.put(topic.getInLongCluster().getClusterId(),
pulsarClient);
-LOGGER.debug("create pulsar client succ {}",
-new
String[]{top
Re: [PR] [INLONG-11129][Sort] Enhanced Metric Instrumentation for InLong Sort Flink Connector [inlong]
PeterZh6 commented on PR #11130: URL: https://github.com/apache/inlong/pull/11130#issuecomment-2363470575 > @PeterZh6 please fix the failed workflow, thanks. Thanks for following up with this. However, I am unable to reproduce the test failure locally, so may still need some time. -- 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
Re: [PR] [INLONG-11148][Sort] Fixes Log4j2 of sort-end-to-end-test-v1.15 not working properly [inlong]
PeterZh6 commented on PR #11149: URL: https://github.com/apache/inlong/pull/11149#issuecomment-2363471108 > @PeterZh6 please fix the failed workflows, thanks. Thanks for following up with this. However, I am unable to reproduce the test failure locally, so may still need some time -- 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
Re: [PR] [INLONG-10464][SDK] InlongSDK support retry sending when failed [inlong]
emptyOVO commented on code in PR #11144:
URL: https://github.com/apache/inlong/pull/11144#discussion_r1768424093
##
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/ConfigConstants.java:
##
@@ -74,4 +74,6 @@ public class ConfigConstants {
public static String HTTP = "http://";;
public static String HTTPS = "https://";;
+public static int SENDER_MAX_RETRY = 3;
Review Comment:
done
--
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
