@Override
  public void open() throws Exception {
    ExecutionConfig.GlobalJobParameters globalJobParameters =
        getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
    String taskId =
        Objects.nonNull(globalJobParameters.toMap().get(INFLUXDB_TAG_NAME))
            ? globalJobParameters.toMap().get(INFLUXDB_TAG_NAME)
            : "null";
    // latency
    if (metricsGenerator.enable()) {
      getRuntimeContext()
          .getMetricGroup()
          .addGroup(INFLUXDB_TAG_NAME, taskId)
          .gauge("record-latency", metricsGenerator::getCurrentLatency);
      LOG.info("add metrics record-latency");
    }
    if (metricsGenerator.isMetricEnable()) {
      // speed
      meterFlowRate =
          getRuntimeContext()
              .getMetricGroup()
              .addGroup(INFLUXDB_TAG_NAME, taskId)
              .meter("record-meter", new MeterView(60));
      LOG.info("add metrics record-meter");
      // rate of flow
      meterSpeed =
          getRuntimeContext()
              .getMetricGroup()
              .addGroup(INFLUXDB_TAG_NAME, taskId)
              .meter("record-count", new MeterView(60));
      LOG.info("add metrics record-count");
    }
    if (logWriter != null) {
      logWriter.open();
    }
    if (fileWriter != null) {
      fileWriter.open();
    }
  }
This is the general entry of an mixed-format writer that wraps different operators insides.

@param <OUT>
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/MixedFormatWriter.java