title: TimeoutException: Timeout of 60000ms expired before the position for partition could be determined when I try to read from kafka to flink
body: <p>I'm trying to read records from a topic in Kafka which is produced by a kafka connect jdbc source connector , here is the connector config:</p>
  <pre><code>&quot;name&quot;: &quot;customers&quot;,
  &quot;config&quot;: {
      &quot;poll.interval.ms&quot;: &quot;3000&quot;,
      &quot;table.poll.interval.ms&quot;: &quot;10000&quot;,
      &quot;errors.log.enable&quot;: &quot;false&quot;,
      &quot;errors.log.include.messages&quot;: &quot;false&quot;,
      &quot;connector.class&quot;: &quot;io.confluent.connect.jdbc.JdbcSourceConnector&quot;,
      &quot;topic.prefix&quot;: &quot;_&quot;,
      &quot;mode&quot;: &quot;incrementing&quot;,
      &quot;validate.non.null&quot;: &quot;true&quot;,
      &quot;table.whitelist&quot;: &quot;customers&quot;,
      &quot;incrementing.column.name&quot;: &quot;customer_id&quot;,
      &quot;connection.url&quot;: &quot;jdbc:sqlserver://demo-sqlserver:1433;databaseName=DemoData&quot;,
      &quot;connection.user&quot;: &quot;sa&quot;,
      &quot;connection.password&quot;: &quot;password&quot;,
      &quot;key.converter&quot;: &quot;io.confluent.connect.avro.AvroConverter&quot;,
      &quot;key.converter.schema.registry.url&quot;: &quot;http://kafka-schema-registry:8081&quot;,
      &quot;value.converter&quot;: &quot;io.confluent.connect.avro.AvroConverter&quot;,
      &quot;value.converter.schema.registry.url&quot;: &quot;http://kafka-schema-registry:8081&quot;,
      &quot;tasks.max&quot;: &quot;2&quot;
  }
  </code></pre>
  <p>I'm using docker-compose to run kafka stack containers with this images:
  zookeeper:3.4.9, confluentinc/cp-kafka:5.5.1, confluentinc/cp-schema-registry:5.5.1, confluentinc/cp-kafka-connect:5.5.1</p>
  <p>and I'm using flink 1.12.0 and here is my docker-compose and docker file for flink:</p>
  <pre><code>services:
  
    jobmanager:
      build:
        dockerfile: Dockerfile
        context: .
      volumes:
        - ./examples:/opt/examples
        - ./opt/flink/usrlib:/opt/flink/usrlib
      hostname: &quot;jobmanager&quot;
      expose:
        - &quot;6123&quot;
      ports:
        - &quot;28081:8081&quot;
      command: jobmanager
      environment:
        - JOB_MANAGER_RPC_ADDRESS=jobmanager
      networks:
        - flink-proxy-net
    taskmanager:
      build:
        dockerfile: Dockerfile
        context: .
      volumes:
        - ./examples:/opt/examples
        - ./opt/flink/usrlib:/opt/flink/usrlib
      expose:
        - &quot;6121&quot;
        - &quot;6122&quot;
      depends_on:
        - jobmanager
      command: taskmanager
      links:
        - jobmanager:jobmanager
      environment:
        - JOB_MANAGER_RPC_ADDRESS=jobmanager
      networks:
        - flink-proxy-net
  </code></pre>
  <p>Dockerfile:</p>
  <pre><code>FROM apache/flink:1.12-scala_2.11-java11
  
  RUN set -ex 
  RUN apt-get update
  RUN apt-get -y install python3
  RUN apt-get -y install python3-pip 
  RUN apt-get -y install python3-dev 
  RUN ln -s /usr/bin/python3 /usr/bin/python 
  RUN ln -s /usr/bin/pip3 /usr/bin/pip
  
  RUN set -ex; 
  RUN apt-get update; 
  RUN python -m pip install --upgrade pip; 
  RUN pip install apache-flink
  
  ARG FLINK_VERSION=1.12.0
  
  RUN wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-json/${FLINK_VERSION}/flink-json-${FLINK_VERSION}.jar
  RUN wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-csv/${FLINK_VERSION}/flink-csv-${FLINK_VERSION}.jar
  RUN wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.12/${FLINK_VERSION}/flink-sql-connector-kafka_2.12-${FLINK_VERSION}.jar
  RUN wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-avro/${FLINK_VERSION}/flink-sql-avro-${FLINK_VERSION}.jar
  
  RUN mkdir -p /opt/data; \
      mkdir -p /opt/data/stream
  
  WORKDIR /opt/flink
  </code></pre>
  <p>I'm trying to run a simple example of reading a topic from kafka and write data to kafka again with this python code:</p>
  <pre><code>from pyflink.datastream import StreamExecutionEnvironment
  from pyflink.table import StreamTableEnvironment, EnvironmentSettings
  
  b_s_env = StreamExecutionEnvironment.get_execution_environment()
  b_s_env.set_parallelism(1)
  b_s_settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
  b_s_t_env = StreamTableEnvironment.create(b_s_env, environment_settings=b_s_settings)
  src_kafka_ddl=&quot;&quot;&quot;
    CREATE TABLE SourceTable (
      `age` INT,
      `customer_id` INT,
      `name` STRING
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'dj_customers',
      'properties.bootstrap.servers' = 'kafka1:9092',
      'properties.zookeeper.connect'='zoo1:2181',
      'scan.startup.mode' = 'earliest-offset',
      'format' = 'avro'
    )
  &quot;&quot;&quot;
  sink_kafka_ddl=&quot;&quot;&quot;
    CREATE TABLE SinkTable (
      `customer_id` INT,
      `name` STRING
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'sink_test',
      'properties.bootstrap.servers' = 'kafka1:9092',
      'properties.zookeeper.connect'='zoo1:2181',
      'scan.startup.mode' = 'earliest-offset',
      'format' = 'avro'
    )
  &quot;&quot;&quot;
  
  b_s_t_env.execute_sql(src_kafka_ddl)
  b_s_t_env.execute_sql(sink_kafka_ddl)
  orders = b_s_t_env.from_path(&quot;SourceTable&quot;) 
  orders.select(orders.customer_id, orders.name).execute_insert(&quot;SinkTable&quot;).wait()
  </code></pre>
  <p>after submitting the job I get timeout exception</p>
  <pre><code>Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition _customers-0 could be determined. 
  
      Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
          at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
          at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:123)
          ... 18 more
      Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
          at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:118)
          at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80)
          at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233)
          at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224)
          at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215)
          at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:665)
          at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89)
          at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:447)
          at jdk.internal.reflect.GeneratedMethodAccessor41.invoke(Unknown Source)
          at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
          at java.base/java.lang.reflect.Method.invoke(Unknown Source)
          at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:306)
          at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:213)
          at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77)
          at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:159)
          at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
          at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
          at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
          at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
          at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
          at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
          at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
          at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
          at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
          at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
          at akka.actor.ActorCell.invoke(ActorCell.scala:561)
          at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
          at akka.dispatch.Mailbox.run(Mailbox.scala:225)
          at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
          at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
          at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
          at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
          at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
      Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition _customers-0 could be determined
  </code></pre>