API Flink Table: невозможно зарегистрировать таблицу в пакетной среде, обменивающейся данными с Kafka

Question

Здравствуйте, уважаемое сообщество stackoverflow,

У меня есть (вероятно, базовая c) проблема с настройкой с использованием Apache Flink Table API с Apache Kafka на BackEnd. Моя цель - прочитать таблицу из файла .csv и переслать данные с тем же форматом и схемой в Apache Kafka. Я использую Eclipse на машине Windows, в то время как Kafka устанавливается на виртуальную машину Ubuntu 18.04 через VMWware. Kafka установлен и работает, как описано здесь .

Я получаю сообщение «Никакая таблица не была зарегистрирована под именем default_catalog. default_database. MyKafkaFlinkTable исключение в последней строке кода, хотя вызов «.from» одной строки ранее, который создает объект «Table» из «TemporaryView» в данном месте («MyKafkaFlinkTable»), не вызвал исключение, поэтому «TemporaryView» существует во время выполнения. Других каталогов нет. или базы данных, зарегистрированные там, где может быть таблица, и функция '.insertInto' может посмотреть.

Кто-нибудь знает, что мне не хватает?

Я использую Maven для управления библиотекой, Пом. xml проекта:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>MyGroupID</groupId>
  <artifactId>MyArtifactID</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>MyArtifactID</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-java</artifactId>
      <version>1.10.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-java-bridge_2.12</artifactId>
      <version>1.10.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner_2.12</artifactId>
      <version>1.10.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-scala_2.12</artifactId>
      <version>1.10.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-uber_2.12</artifactId>
        <version>1.10.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_2.12</artifactId>
        <version>1.10.0</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-runtime-blink_2.12</artifactId>
        <version>1.10.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-uber-blink_2.12</artifactId>
        <version>1.10.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-csv</artifactId>
        <version>1.10.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-optimizer -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-optimizer_2.12</artifactId>
        <version>1.10.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.12</artifactId>
        <version>1.10.0</version>
    </dependency>
    <dependency>
        <groupId>jdk.tools</groupId>
        <artifactId>jdk.tools</artifactId>
        <version>1.8</version>
        <scope>system</scope>
        <systemPath>C:\Program Files\Java\jdk1.8.0_241\lib\tools.jar</systemPath>
    </dependency>
  </dependencies>
</project>

Java -Код выглядит следующим образом:

package MyGroupID.MyArtifactID;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

public class Flink_Kafka {
    public static void main (String [] args) {
        EnvironmentSettings bbSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inBatchMode()
                .build();
        TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);
        DataSet<Tuple2<Integer, String>> csvInput = fbenv.readCsvFile("C:\\test_csv.csv").types(Integer.class,String.class);
        Table tableCsv = fbTableEnv.fromDataSet(csvInput);
        Schema schema = new Schema()
                .field("counter", DataTypes.INT())
                .field("description", DataTypes.STRING());
        bbTableEnv.connect(new Kafka()
                .version("universal")
                .topic("flink_test")
                .property("zookeeper.connect", "<Ip of virtual machine>:2182")
                .property("bootstrap.servers", "<Ip of virtual machine>:9092")
                .property("group.id", "flink_group"))
                .withFormat(new Csv())
                .withSchema(schema)
                .createTemporaryTable("MyKafkaFlinkTable");
        Table table = bbTableEnv.from("default_catalog.default_database.MyKafkaFlinkTable");
        tableCsv.insertInto("MyKafkaFlinkTable");
    }
}

Исключение составляют следующие:

Exception in thread "main" org.apache.flink.table.api.TableException: No table was registered under the name `default_catalog`.`default_database`.`MyKafkaFlinkTable`.
    at org.apache.flink.table.api.internal.TableEnvImpl.insertInto(TableEnvImpl.scala:639)
    at org.apache.flink.table.api.internal.TableEnvImpl.insertInto(TableEnvImpl.scala:607)
    at org.apache.flink.table.api.internal.TableImpl.insertInto(TableImpl.java:411)
    at MyGroupID.MyArtifactID.Flink_Kafka.main(Flink_Kafka.java:56)