5

I have one questions about running Flink streaming job in IDE or as fat jar without deploying it to Flink server.

The problem is I cannot run it in IDE when I have more than 1 taskslot in my job.

public class StreamingJob {

public static void main(String[] args) throws Exception {
    // set up the streaming execution environment
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Properties kafkaProperties = new Properties();
    kafkaProperties.setProperty("bootstrap.servers", "localhost:9092");
    kafkaProperties.setProperty("group.id", "test");
    env.setParallelism(1);

    DataStream<String> kafkaSource = env
        .addSource(new FlinkKafkaConsumer010<>("flink-source", new SimpleStringSchema(), kafkaProperties))
        .name("Kafka-Source")
        .slotSharingGroup("Kafka-Source");

    kafkaSource.print().slotSharingGroup("Print");

    env.execute("Flink Streaming Java API Skeleton");
}

}

I know that job need 2 slot for this job and I can have two taskmanagers in Flink cluster, but how can I run it locally in IDE.

Currently I have to specify the same slotSharingGroup name for all operator locally to have one slot. But it's not flexible.

How do you handle it?

Till Rohrmann
  • 13,148
  • 1
  • 25
  • 51
Aleksandr Filichkin
  • 660
  • 2
  • 8
  • 22

1 Answers1

8

This is a known bug which you are describing. You can find the corresponding JIRA issue here.

The way to circumvent this problem is to manually set the number of task slots with which the TaskExecutor is started. You can do this via the TaskManagerOptions.NUM_TASK_SLOTS configuration option:

final int parallelism = ...;
final Configuration configuration = new Configuration();
configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);

final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(parallelism, configuration);
Till Rohrmann
  • 13,148
  • 1
  • 25
  • 51