content: /*
   * Copyright 2022 The FeatHub Authors
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *     https://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.flink.streaming.connectors.kafka.table;
  
  import org.apache.flink.api.common.restartstrategy.RestartStrategies;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  import org.apache.flink.test.util.AbstractTestBase;
  import org.apache.flink.util.DockerImageVersions;
  
  import org.apache.kafka.clients.CommonClientConfigs;
  import org.apache.kafka.clients.admin.AdminClient;
  import org.apache.kafka.clients.admin.NewTopic;
  import org.junit.Before;
  import org.junit.ClassRule;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.testcontainers.containers.KafkaContainer;
  import org.testcontainers.containers.output.Slf4jLogConsumer;
  import org.testcontainers.utility.DockerImageName;
  
  import java.time.Duration;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Properties;
  
  // TODO: This class will be removed after FLINK-24456 resolved.
  /**
   * Base class for Kafka Table IT Cases.
   *
   * <p>This class is copied from Apache Flink project. We cannot re-use the class as it is not
   * packaged in the test jar.
   */
  public abstract class KafkaTableTestBase extends AbstractTestBase {
      private static final Logger LOG = LoggerFactory.getLogger(KafkaTableTestBase.class);
  
      private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
      private static final int zkTimeoutMills = 30000;
  
      @ClassRule
      public static final KafkaContainer KAFKA_CONTAINER =
              new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA)) {
                  @Override
                  protected void doStart() {
                      super.doStart();
                      if (LOG.isInfoEnabled()) {
                          this.followOutput(new Slf4jLogConsumer(LOG));
                      }
                  }
              }.withEmbeddedZookeeper()
                      .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS)
                      .withEnv(
                              "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",
                              String.valueOf(Duration.ofHours(2).toMillis()))
                      // Disable log deletion to prevent records from being deleted during test run
                      .withEnv("KAFKA_LOG_RETENTION_MS", "-1");
  
      protected StreamExecutionEnvironment env;
      protected StreamTableEnvironment tEnv;
  
      @Before
      public void setup() {
          env = StreamExecutionEnvironment.getExecutionEnvironment();
          tEnv = StreamTableEnvironment.create(env);
          env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
      }
  
      public Properties getStandardProps() {
          Properties standardProps = new Properties();
          standardProps.put("bootstrap.servers", KAFKA_CONTAINER.getBootstrapServers());
          standardProps.put("group.id", "flink-tests");
          standardProps.put("enable.auto.commit", false);
          standardProps.put("auto.offset.reset", "earliest");
          standardProps.put("max.partition.fetch.bytes", 256);
          standardProps.put("zookeeper.session.timeout.ms", zkTimeoutMills);
          standardProps.put("zookeeper.connection.timeout.ms", zkTimeoutMills);
          return standardProps;
      }
  
      public String getBootstrapServers() {
          return KAFKA_CONTAINER.getBootstrapServers();
      }
  
      public void createTestTopic(String topic, int numPartitions, int replicationFactor) {
          Map<String, Object> properties = new HashMap<>();
          properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
          try (AdminClient admin = AdminClient.create(properties)) {
              admin.createTopics(
                      Collections.singletonList(
                              new NewTopic(topic, numPartitions, (short) replicationFactor)));
          }
      }
  
      public void deleteTestTopic(String topic) {
          Map<String, Object> properties = new HashMap<>();
          properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
          try (AdminClient admin = AdminClient.create(properties)) {
              admin.deleteTopics(Collections.singletonList(topic));
          }
      }
  }
  