Thankyou Vidya
Currently, we have a Flink job in the Cloudera CDP 7.1.9 environment with Flink Version: 1.16.2-csa1.11.0.2. We are attempting to create a Flink job to read data from Kafka and send it back to Kafka.
The Kafka configuration we are using is as follows:
kafka.group.id=TEST_mig_2
kafka.flink.partition-discovery.interval-millis=120000
kafka.retries=5
kafka.linger.ms=500
kafka.retry.backoff.ms=100
kafka.compression.type=lz4
kafka.request.timeout.ms=180000
kafka.delivery.timeout.ms=302000
kafka.transaction.timeout.ms=900000
kafka.auto.offset.reset=earliest
kafka.isolation.level=read_committed
and the Kafka writer snapshot we are using is:
public class WriterModule implements Write<String, String> {
@Override
public DataStream<String> write(
DataStream<String> transactionStream, ParameterTool params, StreamExecutionEnvironment env) {
String outputTopic = params.get("app.topic.output");
Properties kafkaProps = Utils.readKafkaPropertiesSink(params);
String appName = params.get("app.name", "Generic");
int sinkParallelism=params.getInt("app.input.sink.parallelism");
KafkaSink<String> streamOutput = KafkaSink.<String>builder()
.setBootstrapServers(params.get(Utils.KAFKA_BOOTSTRAP_SERVERS_SINK))
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic(outputTopic)
.setValueSerializationSchema(new SimpleStringSchema())
.build())
.setKafkaProducerConfig(kafkaProps)
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix(appName+"-"+outputTopic)
.build();
transactionStream.sinkTo(streamOutput).setParallelism(sinkParallelism).name("trxid"+appName+outputTopic+"dup").uid(appName+"-"+outputTopic);
return transactionStream;
}
}
