@Override
    public void processElement(T value) throws Exception {

        try {
            if (disableSwitch()) {
                writer.write((RowData) value);
            } else {
                if (isIncrementalPhase((RowData) value)) {
                    switchToUpsert();
                }
                writer.write(switchHelper.removeIncrementalField((RowData) value));
            }
        } catch (Exception e) {
            if (multipleSink) {
                throw e;
            }

            LOGGER.error(String.format("write error, raw data: %s", value), e);
            if (!dirtyOptions.ignoreDirty()) {
                throw e;
            }
            if (dirtySink != null) {
                DirtyData.Builder<Object> builder = DirtyData.builder();
                if (!disableSwitch()) {
                    value = (T) switchHelper.removeIncrementalField((RowData) value);
                }
                try {
                    builder.setData(value)
                            .setLabels(dirtyOptions.getLabels())
                            .setLogTag(dirtyOptions.getLogTag())
                            .setIdentifier(dirtyOptions.getIdentifier())
                            .setRowType(flinkRowType)
                            .setDirtyMessage(e.getMessage());
                    dirtySink.invoke(builder.build());
                    if (metricData != null) {
                        metricData.invokeDirtyWithEstimate(value);
                    }
                } catch (Exception ex) {
                    if (!dirtyOptions.ignoreSideOutputErrors()) {
                        throw new RuntimeException(ex);
                    }
                    LOGGER.warn("Dirty sink failed", ex);
                }
            }
            return;
        }
        if (metricData != null) {
            metricData.invokeWithEstimate(value);
        }
    }
this method should only be called in open() method
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java