I tried Flink version 1.14.2 / 1.13.5 On Fri, Jan 7, 2022 at 11:46 AM Flink Lover <flinkbyhe...@gmail.com> wrote:
> Also, I am using flink-connector-kafka_2.11 > > val consumer = new FlinkKafkaConsumer[String]("topic_name", new > SimpleStringSchema(), properties) > > > val myProducer = new FlinkKafkaProducer[String]( > "topic_name", // target topic > new KeyedSerializationSchemaWrapper[String](new > SimpleStringSchema()), // serialization schema > getProperties(), // producer config > FlinkKafkaProducer.Semantic.AT_LEAST_ONCE) > > > > On Fri, Jan 7, 2022 at 11:43 AM Flink Lover <flinkbyhe...@gmail.com> > wrote: > >> Hi All, >> >> I checked the Java version using the java -version on the terminal and it >> gave me 1.8.0.281. Also, the project has been compiled using JDK 8 only >> which is by default. >> >> [image: image.png] >> >> What do you mean by target jvm? Also, what I am trying to achieve is >> correct? about the windows? >> >> Thanks, >> Martin >> >> On Fri, Jan 7, 2022 at 8:07 AM Qingsheng Ren <renqs...@gmail.com> wrote: >> >>> Hi Martin, >>> >>> Can you provide the configuration of your Kafka producer and consumer? >>> Also it’ll be helpful to have the complete code of your DataStream. >>> >>> About the error you mentioned, I doubt that the JDK version you actually >>> use is probably below 1.8. Can you have a double check on the environment >>> that your job is running in? >>> >>> Cheers, >>> >>> Qingsheng Ren >>> >>> >>> > On Jan 7, 2022, at 1:13 AM, Flink Lover <flinkbyhe...@gmail.com> >>> wrote: >>> > >>> > Hello Folks! >>> > >>> > I have a DataStream which sends data to the consumer but I got the >>> data once the code completed its execution. I didn't receive the records as >>> the code was writing it to the topic. I was able to achieve this behavior >>> using AT_LEAST_ONCE property but I decided to implement Watermarks. I >>> haven't enabled checkpointing as of now. I know checkpointing will also do >>> the trick. My expectation is Producer should batch the records of 2 >>> seconds and send it to the consumer and consumer should receive a batch of >>> 2 seconds. My code goes as below: >>> > >>> > Producer Side: >>> > dataToKafka.assignTimestampsAndWatermarks( >>> > WatermarkStrategy >>> > .forBoundedOutOfOrderness[String](Duration.ofSeconds(2))) >>> > dataToKafka.addSink(myProducer).uid("source") >>> > >>> > Consumer Side: >>> > consumer.assignTimestampsAndWatermarks( >>> > WatermarkStrategy >>> > .forBoundedOutOfOrderness[String](Duration.ofSeconds(2))) >>> > >>> > Now this gives me an error as below: >>> > >>> > Static methods in interface require -target:jvm-1.8 >>> > .forBoundedOutOfOrderness[String](Duration.ofSeconds(2))) >>> > >>> > My scala version is 2.11.12 and Java JDK 1.8.0.281 >>> > >>> > Thanks, >>> > Martin. >>> > >>> > >>> > >>> >>>