You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "XiaoYou201 (via GitHub)" <gi...@apache.org> on 2024/06/18 08:13:50 UTC
[PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once [inlong]
XiaoYou201 opened a new pull request, #10440:
URL: https://github.com/apache/inlong/pull/10440
### [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once
Fixes #10311
### Motivation
Now，every connectors report audit information only support at least once. So，This pr base it to implements report audit information exactly once. The audit information will not be wrong in much exception situation.
### Modifications
Using the checkpoint principle in Flink to modify the process, the modified flow chart is shown in Figure 1 and Figure 2. In Figure 1, the callback method of notifyCompleteCheckpoint is used to upload audit information instead of scheduled upload.
Each Source/Sink will save the checkpointId of the currently ongoing checkpoint, which is nowCheckpointId in the figure. When the audit information is written to the Buffer, nowCheckpointId will be attached, indicating that the audit information
is written during this ongoing checkpoint. The audit information and checkpointId are in a many-to-one relationship.
When a snapshot request is received, the current operator's nowCheckpointId is updated to (snapshot) checkpointId + 1. When all operators in a task complete the snapshot, the notifyCompleteCheckpoint method is called back.
At this time, AuditBuffer uploads audit information less than or equal to checkpointId (parameters in the notifyCompleteCheckpoint method).
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "XiaoYou201 (via GitHub)" <gi...@apache.org>.
XiaoYou201 commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644283540
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java:
##########
@@ -257,18 +260,24 @@ public void run(SourceContext<T> ctx) throws Exception {
lastConsumeInstant = Instant.now();
List<T> records = new ArrayList<>();
- lastConsumeInstant = getRecords(lastConsumeInstant, messageList, records);
synchronized (ctx.getCheckpointLock()) {
-
+ lastConsumeInstant = getRecords(lastConsumeInstant, messageList, records);
for (T record : records) {
ctx.collect(record);
}
- currentOffsets.put(
- consumeResult.getPartitionKey(),
- consumeResult.getCurrOffset());
+ ConsumerResult confirmResult = confirmOffset(consumeResult);
Review Comment:
already add some description and change the title.
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "XiaoYou201 (via GitHub)" <gi...@apache.org>.
XiaoYou201 commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644284075
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/table/DynamicTubeMQTableDeserializationSchema.java:
##########
@@ -110,9 +110,15 @@ public void deserialize(Message message, Collector<RowData> out) throws IOExcept
}
@Override
- public void flushAudit() {
+ public void flushAuditById(long checkpointId) {
if (sourceMetricData != null) {
- sourceMetricData.flushAuditData();
+ sourceMetricData.flushAuditById(checkpointId);
+ }
+ }
+ @Override
+ public void setNowCheckpointId(long checkpointId) {
+ if (sourceMetricData != null) {
+ sourceMetricData.setNowCheckpointId(checkpointId + 1);
Review Comment:
fix it, thx~
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644349734
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricsCollector.java:
##########
@@ -31,11 +31,10 @@ public class MetricsCollector<T> implements TimestampedCollector<T> {
private long timestampMillis;
- SourceMetricData metricData;
-
+ private final SourceMetricsReporter metricsReporter;
public MetricsCollector(Collector<T> collector,
- SourceMetricData sourceMetricData) {
- this.metricData = sourceMetricData;
+ SourceMetricsReporter sourceMetricData) {
Review Comment:
SourceMetricsReporter class name doesn't match sourceMetricData variable name
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once [inlong]
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644066395
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/table/DynamicTubeMQTableDeserializationSchema.java:
##########
@@ -110,9 +110,15 @@ public void deserialize(Message message, Collector<RowData> out) throws IOExcept
}
@Override
- public void flushAudit() {
+ public void flushAuditById(long checkpointId) {
if (sourceMetricData != null) {
- sourceMetricData.flushAuditData();
+ sourceMetricData.flushAuditById(checkpointId);
+ }
+ }
+ @Override
+ public void setNowCheckpointId(long checkpointId) {
+ if (sourceMetricData != null) {
+ sourceMetricData.setNowCheckpointId(checkpointId + 1);
Review Comment:
method name doesn't match the implementation, added id by 1 but the method name is set
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "XiaoYou201 (via GitHub)" <gi...@apache.org>.
XiaoYou201 commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644400805
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricsCollector.java:
##########
@@ -31,11 +31,10 @@ public class MetricsCollector<T> implements TimestampedCollector<T> {
private long timestampMillis;
- SourceMetricData metricData;
-
+ private final SourceMetricsReporter metricsReporter;
public MetricsCollector(Collector<T> collector,
- SourceMetricData sourceMetricData) {
- this.metricData = sourceMetricData;
+ SourceMetricsReporter sourceMetricData) {
Review Comment:
fix it， thx～
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java:
##########
@@ -311,14 +324,21 @@ private Instant getRecords(Instant lastConsumeInstant, List<Message> messageList
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
-
+ deserializationSchema.setCurrentCheckpointId(context.getCheckpointId());
+ // Make sure snapshot all tube partitions' offset to avoid inconsistency. See BUG-124774267 for details.
Review Comment:
fix it， thx～
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1645284647
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricsCollector.java:
##########
@@ -31,11 +31,10 @@ public class MetricsCollector<T> implements TimestampedCollector<T> {
private long timestampMillis;
- SourceMetricData metricData;
-
+ private final SourceMetricsReporter metricsReporter;
public MetricsCollector(Collector<T> collector,
- SourceMetricData sourceMetricData) {
- this.metricData = sourceMetricData;
+ SourceMetricsReporter sourceMetricData) {
Review Comment:
no change 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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644354061
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceExactlyMetric.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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.metric;
+
+import org.apache.inlong.audit.AuditReporterImpl;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SimpleCounter;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.inlong.audit.consts.ConfigConstants.DEFAULT_AUDIT_TAG;
+import static org.apache.inlong.common.constant.Constants.DEFAULT_AUDIT_VERSION;
+import static org.apache.inlong.sort.base.Constants.CURRENT_EMIT_EVENT_TIME_LAG;
+import static org.apache.inlong.sort.base.Constants.CURRENT_FETCH_EVENT_TIME_LAG;
+import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN;
+import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN_FOR_METER;
+import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN_PER_SECOND;
+import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_IN;
+import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_IN_FOR_METER;
+import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_IN_PER_SECOND;
+import static org.apache.inlong.sort.base.util.CalculateObjectSizeUtils.getDataSize;
+
+public class SourceExactlyMetric implements MetricData, Serializable, SourceMetricsReporter {
Review Comment:
some of the methods are exactly the same as in sourceMetricData, please extract a base class or abstract class
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java:
##########
@@ -311,14 +324,21 @@ private Instant getRecords(Instant lastConsumeInstant, List<Message> messageList
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
-
+ deserializationSchema.setCurrentCheckpointId(context.getCheckpointId());
+ // Make sure snapshot all tube partitions' offset to avoid inconsistency. See BUG-124774267 for details.
Review Comment:
what is BUG-124774267 ?
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once [inlong]
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644063097
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java:
##########
@@ -257,18 +260,24 @@ public void run(SourceContext<T> ctx) throws Exception {
lastConsumeInstant = Instant.now();
List<T> records = new ArrayList<>();
- lastConsumeInstant = getRecords(lastConsumeInstant, messageList, records);
synchronized (ctx.getCheckpointLock()) {
-
+ lastConsumeInstant = getRecords(lastConsumeInstant, messageList, records);
for (T record : records) {
ctx.collect(record);
}
- currentOffsets.put(
- consumeResult.getPartitionKey(),
- consumeResult.getCurrOffset());
+ ConsumerResult confirmResult = confirmOffset(consumeResult);
Review Comment:
please add description on how this solve the tube message duplicate ( not audit)
and change the title make it contains this fix
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "XiaoYou201 (via GitHub)" <gi...@apache.org>.
XiaoYou201 commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644285163
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/table/DynamicTubeMQTableDeserializationSchema.java:
##########
@@ -110,9 +110,15 @@ public void deserialize(Message message, Collector<RowData> out) throws IOExcept
}
@Override
- public void flushAudit() {
+ public void flushAuditById(long checkpointId) {
if (sourceMetricData != null) {
- sourceMetricData.flushAuditData();
+ sourceMetricData.flushAuditById(checkpointId);
+ }
+ }
+ @Override
+ public void setNowCheckpointId(long checkpointId) {
+ if (sourceMetricData != null) {
+ sourceMetricData.setNowCheckpointId(checkpointId + 1);
Review Comment:
already change the implement method, thx~
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "XiaoYou201 (via GitHub)" <gi...@apache.org>.
XiaoYou201 commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644282617
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricsCollector.java:
##########
@@ -31,11 +31,10 @@ public class MetricsCollector<T> implements TimestampedCollector<T> {
private long timestampMillis;
- SourceMetricData metricData;
-
+ SourceMetricsReporter metricsReporter;
Review Comment:
fix it, thx~
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "XiaoYou201 (via GitHub)" <gi...@apache.org>.
XiaoYou201 commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1645294317
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricsCollector.java:
##########
@@ -31,11 +31,10 @@ public class MetricsCollector<T> implements TimestampedCollector<T> {
private long timestampMillis;
- SourceMetricData metricData;
-
+ private final SourceMetricsReporter metricsReporter;
public MetricsCollector(Collector<T> collector,
- SourceMetricData sourceMetricData) {
- this.metricData = sourceMetricData;
+ SourceMetricsReporter sourceMetricData) {
Review Comment:
fix it， thx～
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once [inlong]
Posted by "vernedeng (via GitHub)" <gi...@apache.org>.
vernedeng commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644068341
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/table/DynamicTubeMQTableDeserializationSchema.java:
##########
@@ -110,9 +110,15 @@ public void deserialize(Message message, Collector<RowData> out) throws IOExcept
}
@Override
- public void flushAudit() {
+ public void flushAuditById(long checkpointId) {
if (sourceMetricData != null) {
- sourceMetricData.flushAuditData();
+ sourceMetricData.flushAuditById(checkpointId);
+ }
+ }
+ @Override
+ public void setNowCheckpointId(long checkpointId) {
+ if (sourceMetricData != null) {
+ sourceMetricData.setNowCheckpointId(checkpointId + 1);
Review Comment:
why +1 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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once And fix TubeMQ data twice [inlong]
Posted by "XiaoYou201 (via GitHub)" <gi...@apache.org>.
XiaoYou201 commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644433532
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceExactlyMetric.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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.metric;
+
+import org.apache.inlong.audit.AuditReporterImpl;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SimpleCounter;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.inlong.audit.consts.ConfigConstants.DEFAULT_AUDIT_TAG;
+import static org.apache.inlong.common.constant.Constants.DEFAULT_AUDIT_VERSION;
+import static org.apache.inlong.sort.base.Constants.CURRENT_EMIT_EVENT_TIME_LAG;
+import static org.apache.inlong.sort.base.Constants.CURRENT_FETCH_EVENT_TIME_LAG;
+import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN;
+import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN_FOR_METER;
+import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN_PER_SECOND;
+import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_IN;
+import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_IN_FOR_METER;
+import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_IN_PER_SECOND;
+import static org.apache.inlong.sort.base.util.CalculateObjectSizeUtils.getDataSize;
+
+public class SourceExactlyMetric implements MetricData, Serializable, SourceMetricsReporter {
Review Comment:
I will make all connectors support report audit information exactly once in the future, so the sourcemetricdata will be deprecated later
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once [inlong]
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644063097
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java:
##########
@@ -257,18 +260,24 @@ public void run(SourceContext<T> ctx) throws Exception {
lastConsumeInstant = Instant.now();
List<T> records = new ArrayList<>();
- lastConsumeInstant = getRecords(lastConsumeInstant, messageList, records);
synchronized (ctx.getCheckpointLock()) {
-
+ lastConsumeInstant = getRecords(lastConsumeInstant, messageList, records);
for (T record : records) {
ctx.collect(record);
}
- currentOffsets.put(
- consumeResult.getPartitionKey(),
- consumeResult.getCurrOffset());
+ ConsumerResult confirmResult = confirmOffset(consumeResult);
Review Comment:
please add description on how this solve the tube message duplicate ( not audit)
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-10311][Sort] Implement TubeMQ Source report audit information exactly once [inlong]
Posted by "vernedeng (via GitHub)" <gi...@apache.org>.
vernedeng commented on code in PR #10440:
URL: https://github.com/apache/inlong/pull/10440#discussion_r1644041511
##########
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricsCollector.java:
##########
@@ -31,11 +31,10 @@ public class MetricsCollector<T> implements TimestampedCollector<T> {
private long timestampMillis;
- SourceMetricData metricData;
-
+ SourceMetricsReporter metricsReporter;
Review Comment:
private
--
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-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org