org.apache.kafka.common.errors.InvalidGroupIdException

user_12639241 0
Viewed 239

Iv got this error every time I run my simple flink job

org.apache.kafka.common.errors.InvalidGroupIdException: To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.

I tried to add ConsumerConfig.GROUP_ID_CONFIG, it did not work for me

could anyone help me please

here is the code

        StreamExecutionEnvironment cdcEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        cdcEnv.setParallelism(1);

        cdcEnv.enableCheckpointing(5000);
        cdcEnv.setStateBackend(new FsStateBackend(ResourceUtil.getKey("checkpointUri") + "/flink/checkpoint/cp2"));
        cdcEnv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        cdcEnv.getCheckpointConfig().setCheckpointTimeout(5000);
        cdcEnv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        cdcEnv.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", ResourceUtil.getKey("bootstrapServers"));
        properties.setProperty("ConsumerConfig.GROUP_ID_CONFIG", "cloudera_mirrormaker");
        DataStream<String> cdcDataStream = cdcEnv.addSource(new FlinkKafkaConsumer<String>("ods_decl", new SimpleStringSchema(), properties));
        cdcDataStream
                .addSink(new CdcSink(1, 1000L))
                .name("cdc sink.");

        cdcEnv.execute("CDC Flink Job");
apache-kafka
1 Answers

Remove the quotes from "ConsumerConfig.GROUP_ID_CONFIG": replace properties.setProperty("ConsumerConfig.GROUP_ID_CONFIG", "cloudera_mirrormaker"); with properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "cloudera_mirrormaker");

This is because in Java text under double quotes is interpreted as a String.

user_145646850
Related