java/feathub-connectors/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/BoundedKafkaTableITCase.java
<reponame>alibaba/feathub/*
 * 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.streaming.api.datastream.DataStream;
import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;

import org.junit.Test;

import java.util.List;

import static org.assertj.core.api.Assertions.assertThat;

/** Test for {@link BoundedKafkaDynamicSource}. */
public class BoundedKafkaTableITCase extends KafkaTableTestBase {

    @Test
    public void testBoundedKafkaSource() throws Exception {
        final String topic = "test-topic";
        createTestTopic(topic, 1, 1);

        // ---------- Produce an event time stream into Kafka -------------------
        String groupId = getStandardProps().getProperty("group.id");
        String bootstraps = getBootstrapServers();

        final String createTable =
                String.format(
                        "create table kafka (\n"
                                + "  id INTEGER,\n"
                                + "  val DOUBLE\n"
                                + ") with (\n"
                                + "  'connector' = '%s',\n"
                                + "  'topic' = '%s',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'properties.group.id' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  'format' = 'csv'\n"
                                + ")",
                        BoundedKafkaDynamicTableFactory.IDENTIFIER, topic, bootstraps, groupId);

        tEnv.executeSql(createTable);

        String initialValues =
                "INSERT INTO kafka\n"
                        + "SELECT id, val \n"
                        + "FROM (VALUES (1, 2.02), (1, 1.11), (1, 50), (1, 3.1), (2, 5.33), (2, 0.0))\n"
                        + "  AS orders (id, val)";
        tEnv.executeSql(initialValues).await();

        // ---------- Consume stream from Kafka -------------------

        String query = "SELECT * FROM kafka\n";

        DataStream<Row> result = tEnv.toDataStream(tEnv.sqlQuery(query));
        final List<Row> rows = CollectionUtil.iteratorToList(result.executeAndCollect());
        assertThat(rows).hasSize(6);

        // ------------- cleanup -------------------

        deleteTestTopic(topic);
    }
}
