EMsnap commented on code in PR #7716:
URL: https://github.com/apache/inlong/pull/7716#discussion_r1152781708
##########
inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/RecordWithSchema.java:
##########
@@ -56,6 +58,16 @@ public RecordWithSchema(
this.primaryKeys = primaryKeys;
}
+ @Setter
Review Comment:
can be replaced by @Data annotation on RecordWithSchema?
##########
inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergMultipleStreamWriter.java:
##########
@@ -246,55 +246,64 @@ public void processElement(RecordWithSchema
recordWithSchema) throws Exception {
}
if (multipleWriters.get(tableId) != null) {
- for (RowData data : recordWithSchema.getData()) {
+ if (recordWithSchema.isDirty()) {
String dataBaseName = tableId.namespace().toString();
String tableName = tableId.name();
- long size = data == null ? 0 :
data.toString().getBytes(StandardCharsets.UTF_8).length;
+ if (sinkMetricData != null) {
+ sinkMetricData.outputDirtyMetrics(dataBaseName,
+ tableName, recordWithSchema.getRowCount(),
recordWithSchema.getRowSize());
+ }
+ } else {
+ for (RowData data : recordWithSchema.getData()) {
+ String dataBaseName = tableId.namespace().toString();
+ String tableName = tableId.name();
+ long size = data == null ? 0 :
data.toString().getBytes(StandardCharsets.UTF_8).length;
- try {
- multipleWriters.get(tableId).processElement(data);
- } catch (Exception e) {
- LOG.error(String.format("write error, raw data: %s",
data), e);
- if (!dirtyOptions.ignoreDirty()) {
- throw e;
- }
- if (dirtySink != null) {
- DirtyData.Builder<Object> builder =
DirtyData.builder();
- try {
- String dirtyLabel =
DirtySinkHelper.regexReplace(dirtyOptions.getLabels(),
- DirtyType.BATCH_LOAD_ERROR, null,
- dataBaseName, tableName, null);
- String dirtyLogTag =
-
DirtySinkHelper.regexReplace(dirtyOptions.getLogTag(),
- DirtyType.BATCH_LOAD_ERROR, null,
- dataBaseName, tableName, null);
- String dirtyIdentifier =
-
DirtySinkHelper.regexReplace(dirtyOptions.getIdentifier(),
- DirtyType.BATCH_LOAD_ERROR, null,
- dataBaseName, tableName, null);
- builder.setData(data)
- .setLabels(dirtyLabel)
- .setLogTag(dirtyLogTag)
- .setIdentifier(dirtyIdentifier)
-
.setRowType(multipleWriters.get(tableId).getFlinkRowType())
- .setDirtyMessage(e.getMessage());
- dirtySink.invoke(builder.build());
- if (sinkMetricData != null) {
-
sinkMetricData.outputDirtyMetricsWithEstimate(dataBaseName,
- tableName, 1, size);
- }
- } catch (Exception ex) {
- if (!dirtyOptions.ignoreSideOutputErrors()) {
- throw new RuntimeException(ex);
+ try {
+ multipleWriters.get(tableId).processElement(data);
+ } catch (Exception e) {
+ LOG.error(String.format("write error, raw data: %s",
data), e);
+ if (!dirtyOptions.ignoreDirty()) {
+ throw e;
+ }
+ if (dirtySink != null) {
+ DirtyData.Builder<Object> builder =
DirtyData.builder();
+ try {
+ String dirtyLabel =
DirtySinkHelper.regexReplace(dirtyOptions.getLabels(),
Review Comment:
would be better if the construction for dirty data can be extraced since the
method processElement is too long for reading
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
