@Override
    public void open(Configuration parameters) throws Exception {
        this.catalog = catalogLoader.loadCatalog();
        this.multipleWriters = new HashMap<>();
        this.multipleTables = new HashMap<>();
        this.multipleSchemas = new HashMap<>();

        this.runtimeContext = getRuntimeContext();
        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(inlongMetric)
                .withAuditAddress(auditHostAndPorts)
                .withInitRecords(metricState != null ? metricState.getMetricValue(NUM_RECORDS_OUT) : 0L)
                .withInitBytes(metricState != null ? metricState.getMetricValue(NUM_BYTES_OUT) : 0L)
                .withInitDirtyRecords(metricState != null ? metricState.getMetricValue(DIRTY_RECORDS_OUT) : 0L)
                .withInitDirtyBytes(metricState != null ? metricState.getMetricValue(DIRTY_BYTES_OUT) : 0L)
                .withRegisterMetric(MetricOption.RegisteredMetric.ALL)
                .withAuditKeys(auditKeys)
                .build();
        if (metricOption != null) {
            sinkMetricData = new SinkTableMetricData(metricOption, runtimeContext.getMetricGroup());
            sinkMetricData.registerSubMetricsGroup(metricState);
        }

        if (dirtySink != null) {
            try {
                dirtySink.open(new Configuration());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
Iceberg writer that can distinguish different sink tables and route and distribute data into different
IcebergStreamWriter.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergMultipleStreamWriter.java