Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-20157

SourceCoordinatorProvider kills JobManager with IllegalStateException on job submission

    XMLWordPrintableJSON

Details

    • Bug
    • Status: Closed
    • Blocker
    • Resolution: Fixed
    • 1.12.0
    • 1.12.0
    • Connectors / Kafka
    • None

    Description

      While setting up a test job using the new Kafka source for testing the RC1 of Flink 1.12, my JobManager died with a fatal exception:

      2020-11-13 17:05:53,947 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Deploying Flat Map -> Sink: Print to Std. Out (1/1) (attempt #0) with attempt id fc36327d85e775204e82fc8507bf4264 to 192.168.1.25:57387-78ca68 @ robertsbabamac2.localdomain (dataPort=57390) with allocation id a8d918c0cfb57305908ce5a4f4787034
      2020-11-13 17:05:53,988 ERROR org.apache.flink.runtime.util.FatalExitExceptionHandler      [] - FATAL: Thread 'SourceCoordinator-Source: Kafka Source' produced an uncaught exception. Stopping the process...
      java.lang.IllegalStateException: Should never happen. This factory should only be used by a SingleThreadExecutor.
              at org.apache.flink.runtime.source.coordinator.SourceCoordinatorProvider$CoordinatorExecutorThreadFactory.newThread(SourceCoordinatorProvider.java:94) ~[flink-dist_2.11-1.12.0.jar:1.12.0]
              at java.util.concurrent.ThreadPoolExecutor$Worker.<init>(ThreadPoolExecutor.java:619) ~[?:1.8.0_222]
              at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:932) ~[?:1.8.0_222]
              at java.util.concurrent.ThreadPoolExecutor.processWorkerExit(ThreadPoolExecutor.java:1025) ~[?:1.8.0_222]
              at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167) ~[?:1.8.0_222]
              at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_222]
              at java.lang.Thread.run(Thread.java:748) [?:1.8.0_222]
      

      I'm using the KafkaSource as documented, with a single partition topic:

      		KafkaSource<Event> source = KafkaSource
      			 .<Event>builder()
      			 .setBootstrapServers(brokers)
      			 .setGroupId("myGroup")
      			 .setTopics(Arrays.asList(kafkaTopic))
      			 .setDeserializer(new NewEventDeserializer())
      			 .build();
      

      Attachments

        Issue Links

          Activity

            People

              becket_qin Jiangjie Qin
              rmetzger Robert Metzger
              Votes:
              0 Vote for this issue
              Watchers:
              6 Start watching this issue

              Dates

                Created:
                Updated:
                Resolved: