public static KafkaContainer createKafkaContainer(
      String dockerImageVersion, Logger logger, String loggerPrefix) {
    String logLevel;
    if (logger.isTraceEnabled()) {
      logLevel = "TRACE";
    } else if (logger.isDebugEnabled()) {
      logLevel = "DEBUG";
    } else if (logger.isInfoEnabled()) {
      logLevel = "INFO";
    } else if (logger.isWarnEnabled()) {
      logLevel = "WARN";
    } else if (logger.isErrorEnabled()) {
      logLevel = "ERROR";
    } else {
      logLevel = "OFF";
    }

    Slf4jLogConsumer logConsumer = new Slf4jLogConsumer(logger);
    if (!StringUtils.isNullOrWhitespaceOnly(loggerPrefix)) {
      logConsumer.withPrefix(loggerPrefix);
    }
    return new KafkaContainer(DockerImageName.parse(dockerImageVersion))
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
        .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
        .withEnv("KAFKA_LOG4J_ROOT_LOGLEVEL", logLevel)
        .withEnv("KAFKA_LOG4J_LOGGERS", "state.change.logger=" + logLevel)
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
        .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
        .withEnv("KAFKA_TRANSACTION_MAX_TIMEOUT_MS", String.valueOf(Duration.ofHours(2).toMillis()))
        .withEnv("KAFKA_LOG4J_TOOLS_ROOT_LOGLEVEL", logLevel)
        .withLogConsumer(logConsumer);
  }
This method helps to set commonly used Kafka configurations and aligns the internal Kafka log
levels with the ones used by the capturing logger, and set the prefix of logger.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/kafka/testutils/KafkaUtil.java