Realtime Compute for Apache Flink:Kafka connector

Ikhtisar

Prasyarat

Catatan

Tulisan transaksional tidak direkomendasikan karena adanya keterbatasan desain yang diketahui pada Apache Flink dan Apache Kafka. Saat Anda mengatur sink.delivery-guarantee = 'exactly-once', Kafka connector mengaktifkan tulisan transaksional, dengan isu-isu berikut:

• Setiap checkpoint menghasilkan ID Transaksi baru. Jika interval checkpoint terlalu singkat, banjir ID Transaksi yang dihasilkan dapat menyebabkan koordinator kluster Kafka kehabisan memori, sehingga mengganggu stabilitas kluster.

• Setiap transaksi membuat instans Producer baru. Jika terlalu banyak transaksi dikomit secara bersamaan, Pengelola Tugas (TaskManager) dapat kehabisan memori, sehingga mengganggu pekerjaan Apache Flink.

• Jika beberapa pekerjaan Apache Flink menggunakan sink.transactional-id-prefix yang sama, ID Transaksi yang dihasilkan dapat bentrok. Ketika operasi tulis gagal pada satu pekerjaan, hal ini dapat mencegah Log Start Offset (LSO) partisi Apache Kafka maju. Hal ini memengaruhi semua konsumen partisi tersebut.

Jika Anda memerlukan semantik tepat-sekali, gunakan connector Upsert Kafka untuk menulis ke tabel kunci primer, sehingga memastikan idempotensi. Jika Anda harus menggunakan tulisan transaksional, lihat Catatan penggunaan semantik tepat-sekali.

Pemecahan masalah konektivitas jaringan

Error Timed out waiting for a node assignment saat pekerjaan Realtime Compute for Apache Flink gagal dimulai biasanya menunjukkan adanya masalah konektivitas jaringan antara Realtime Compute for Apache Flink dan kluster Kafka.

SQL

Gunakan Kafka connector sebagai tabel sumber atau tabel sink dalam pekerjaan SQL.

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

Kolom metadata

CREATE TABLE kafka_source (
  -- Baca topik pesan sebagai kolom `record_topic`
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Baca timestamp dari ConsumerRecord sebagai kolom `ts`
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Baca offset pesan sebagai kolom `record_offset`
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- Tulis timestamp dari kolom `ts` sebagai timestamp ProducerRecord ke Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

Opsi konektor

• Umum

  Opsi	Deskripsi	Tipe	Wajib	Default	Keterangan
  connector	Jenis connector.	String	Ya	–	Nilainya harus kafka.
  properties.bootstrap.servers	Daftar alamat broker Kafka.	String	Ya	–	Format: host1:port1,host2:port2,.... Pisahkan alamat dengan koma (,).
  properties.*	Properti tambahan untuk klien Kafka.	String	Tidak	–	Kunci properti harus merupakan opsi yang valid sebagaimana didefinisikan dalam dokumentasi resmi Apache Kafka untuk Producer Configs dan Consumer Configs. Realtime Compute for Apache Flink menghapus prefiks properties. dan meneruskan pasangan kunci-nilai yang tersisa ke klien Kafka yang mendasari. Misalnya, Anda dapat mengatur 'properties.allow.auto.create.topics' = 'false' untuk menonaktifkan pembuatan topik otomatis. Kafka connector menimpa opsi-opsi berikut, sehingga Anda tidak dapat mengonfigurasinya dengan cara ini: key.deserializer value.deserializer
  format	Format untuk serialisasi dan deserialisasi nilai pesan Kafka.	String	Tidak	–	Format yang didukung: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Untuk informasi selengkapnya, lihat Opsi format.
  key.format	Format untuk serialisasi dan deserialisasi kunci pesan Kafka.	String	Tidak	–	Format yang didukung: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Catatan Saat Anda menggunakan konfigurasi ini, key.options wajib diisi.
  key.fields	Bidang dari skema tabel yang digunakan sebagai kunci pesan Kafka.	String	Tidak	–	Pisahkan beberapa nama bidang dengan titik koma (;). Contoh: 'field1;field2'.
  key.fields-prefix	Awalan kustom untuk semua bidang kunci guna mencegah bentrok nama dengan bidang nilai.	String	Tidak	–	Awalan ini digunakan untuk membedakan antara bidang kunci dan bidang nilai. Awalan ini dihapus sebelum serialisasi kunci atau setelah deserialisasi. Catatan Jika Anda menggunakan opsi ini, value.fields-include harus diatur ke EXCEPT_KEY.
  value.format	Format untuk serialisasi dan deserialisasi nilai pesan Kafka.	String	Tidak	–	Konfigurasi ini setara dengan format. Anda hanya dapat mengatur salah satu dari format atau value.format. Jika keduanya dikonfigurasi, value.format menggantikan format.
  value.fields-include	Menentukan apakah bidang kunci disertakan dalam format nilai.	String	Tidak	ALL	Nilai yang valid: ALL: Nilai pesan Kafka mencakup semua kolom tabel. EXCEPT_KEY: Nilai pesan Kafka mencakup semua kolom tabel kecuali yang didefinisikan dalam key.fields.

• Tabel sumber

  Opsi	Deskripsi	Tipe	Wajib	Default	Keterangan
  topic	Topik atau topik-topik yang akan dibaca.	String	Tidak	–	Untuk berlangganan ke beberapa topik, pisahkan nama-namanya dengan titik koma (;), contohnya 'topic-1;topic-2'. Catatan Anda hanya dapat menentukan salah satu dari opsi ini atau topic-pattern, tidak keduanya.
  topic-pattern	Ekspresi reguler yang cocok dengan topik yang akan dilanggan. Konsumen berlangganan ke semua topik yang namanya sesuai dengan pola ini.	String	Tidak	–	Contoh: user_event_.*: Cocok dengan semua topik yang diawali dengan user_event_. prod\.logs\..*: Cocok dengan topik yang diawali dengan prod.logs. (karakter . harus di-escape). Catatan Anda hanya dapat menentukan salah satu dari opsi ini atau topic, tidak keduanya.
  properties.group.id	ID kelompok konsumen sumber Kafka.	String	Tidak	KafkaSource-{Nama-Tabel-Sumber}	Jika Anda menggunakan ID kelompok konsumen untuk pertama kalinya, Anda juga harus mengatur properties.auto.offset.reset ke earliest atau latest untuk menentukan offset startup awal.
  scan.startup.mode	Offset startup konsumen Kafka.	String	Tidak	group-offsets	Nilai yang valid: earliest-offset: Mulai membaca dari offset paling awal yang tersedia. latest-offset: Mulai membaca dari offset terbaru. group-offsets: Mulai membaca dari offset yang telah dikomit oleh properties.group.id yang ditentukan. timestamp: Mulai membaca dari scan.startup.timestamp-millis yang ditentukan. specific-offsets: Mulai membaca dari offset yang ditentukan dalam scan.startup.specific-offsets. Catatan Opsi ini hanya berlaku saat pekerjaan dimulai tanpa status. Jika pekerjaan dilanjutkan dari checkpoint, pekerjaan membaca dari offset yang disimpan dalam status checkpoint.
  scan.startup.specific-offsets	Offset awal per-partisi saat scan.startup.mode bernilai specific-offsets.	String	Tidak	–	Contoh: partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	Timestamp awal dalam milidetik saat scan.startup.mode diatur ke timestamp.	Long	Tidak	–	Unitnya adalah milidetik.
  scan.topic-partition-discovery.interval	Interval penemuan partisi.	Durasi	Tidak	5 menit	Connector secara berkala menemukan dan membaca dari partisi baru. Saat Anda menggunakan topic-pattern, connector juga menemukan topik baru yang sesuai dengan pola tersebut. Atur interval ke nilai non-positif untuk menonaktifkan fitur ini. Catatan Di Ververica Runtime (VVR) 6.0.x, penemuan partisi dinamis dinonaktifkan secara default. Mulai dari VVR 8.0, fitur ini diaktifkan secara default dengan interval penemuan 5 menit.
  scan.header-filter	Memfilter pesan berdasarkan header pesan Kafka.	String	Tidak	–	Kunci header dan nilainya dipisahkan dengan tanda titik dua (:). Beberapa kondisi header dihubungkan dengan operator logika (& dan |). Operator logika NOT (!) juga didukung. Contoh: depart:toy|depart:book&!env:test menyimpan data Kafka jika header berisi depart=toy atau depart=book dan tidak berisi env=test. Catatan Opsi ini hanya didukung di Ververica Runtime (VVR) 8.0.6 dan versi setelahnya. Tanda kurung dalam ekspresi tidak didukung. Operasi logika dievaluasi dari kiri ke kanan. Nilai header dikonversi ke string UTF-8 untuk perbandingan.
  scan.check.duplicated.group.id	Memeriksa apakah sudah ada konsumen aktif lain yang menggunakan properties.group.id.	Boolean	Tidak	false	Nilai yang valid: true: Sebelum memulai pekerjaan, sistem memeriksa duplikasi kelompok konsumen. Jika ditemukan, pekerjaan gagal untuk mencegah konflik. false: Memulai pekerjaan tanpa memeriksa konflik. Catatan Opsi ini hanya didukung di Ververica Runtime (VVR) 6.0.4 dan versi setelahnya.

• Tabel sink

  Opsi	Deskripsi	Type	Wajib	Default	Keterangan
  topic	Topik target.	String	Ya	–	–
  sink.partitioner	Memetakan catatan dari instans sink paralel ke partisi Kafka.	String	Tidak	default	Nilai yang valid: default: Menggunakan partitioner Kafka default. fixed: Setiap instans sink paralel menulis ke partisi Kafka tetap. round-robin: Catatan didistribusikan ke partisi secara bergiliran (round-robin). Partitioner kustom: Untuk menggunakan partitioner kustom, berikan nama kelas lengkap dari subclass FlinkKafkaPartitioner, misalnya org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Jaminan pengiriman sink.	String	Tidak	at-least-once	Nilai yang valid: none: Tidak memberikan jaminan apa pun. Catatan mungkin hilang atau diduplikasi. at-least-once: Menjamin tidak ada catatan yang hilang, tetapi mungkin diduplikasi. exactly-once: Menggunakan transaksi Kafka untuk memberikan semantik tepat-sekali, memastikan catatan tidak hilang maupun diduplikasi. Catatan Saat menggunakan semantik exactly-once, Anda juga harus menentukan sink.transactional-id-prefix.
  sink.transactional-id-prefix	Awalan ID transaksi. Diperlukan saat sink.delivery-guarantee bernilai exactly-once.	String	Ya, jika sink.delivery-guarantee bernilai exactly-once	–	Hanya diperlukan saat sink.delivery-guarantee diatur ke exactly-once.
  sink.parallelism	Paralelisme operator sink.	Integer	Tidak	–	Secara default, framework menentukan paralelisme berdasarkan operator hulu.

Keamanan dan otentikasi

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /* Konfigurasi SSL */
  /* Path ke truststore untuk sertifikat CA server. */
  /* File yang diunggah menggunakan Artifacts disimpan di direktori /flink/usrlib/. */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* Jika otentikasi klien diperlukan, Anda juga harus mengonfigurasi path ke keystore (kunci privat). */
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /* Algoritma yang digunakan untuk memverifikasi hostname server. String kosong menonaktifkan verifikasi hostname. */
  'properties.ssl.endpoint.identification.algorithm' = '',
  /* Konfigurasi SASL */
  /* Atur mekanisme SASL ke SCRAM-SHA-256. */
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /* Konfigurasi JAAS. */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

Anda dapat menggunakan fitur Artifacts di Konsol Realtime Compute for Apache Flink untuk mengunggah sertifikat CA dan kunci privat yang disebutkan dalam contoh. File yang diunggah disimpan di direktori /flink/usrlib. Untuk menggunakan file sertifikat CA bernama my-truststore.jks, Anda dapat mengatur properti 'properties.ssl.truststore.location' dalam klausa WITH dengan salah satu dari dua cara berikut:

• Atur 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'. Metode ini menghindari pengunduhan file secara dinamis dari Object Storage Service (OSS) saat runtime, tetapi tidak mendukung Mode Debug.

• Jika versi engine Realtime Compute adalah VVR 11.5 atau versi setelahnya, Anda dapat mengonfigurasi properties.ssl.truststore.location dan properties.ssl.keystore.location ke path OSS absolut. Format path file adalah oss://flink-fullymanaged-<ID Ruang Kerja>/artifacts/namespaces/<Nama Namespace>/<nama file>. Metode ini mengunduh file OSS secara dinamis selama runtime Flink dan mendukung Mode Debug.

Offset awal tabel sumber

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Konsumsi dari offset paling awal.
  'scan.startup.mode' = 'earliest-offset',
  -- Konsumsi dari offset terbaru.
  'scan.startup.mode' = 'latest-offset',
  -- Konsumsi dari offset yang telah dikomit oleh kelompok konsumen "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- Jika "my-group" digunakan pertama kali, konsumsi dimulai dari offset paling awal.
  'properties.auto.offset.reset' = 'latest', -- Jika "my-group" digunakan pertama kali, konsumsi dimulai dari offset terbaru.
  -- Konsumsi dari timestamp tertentu dalam milidetik: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Konsumsi dari offset spesifik.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Mengommit offset sumber

Tabel sumber Kafka hanya mengommit offset konsumennya ke kluster Kafka setelah checkpoint berhasil, sehingga interval checkpoint yang panjang menyebabkan keterlambatan offset yang dikomit. Tabel sumber menyimpan progres bacaan aktual dalam status checkpoint, yang digunakan sistem untuk pemulihan kesalahan. Offset yang dikomit hanya berfungsi sebagai monitor progres dan tidak digunakan untuk pemulihan, sehingga kegagalan komit tidak memengaruhi akurasi data.

Partitioner sink kustom

Jika strategi partisi bawaan Kafka tidak memenuhi kebutuhan Anda, Anda dapat mengimplementasikan partitioner kustom dengan memperluas kelas FlinkKafkaPartitioner. Setelah pengembangan selesai, kompilasi kode Anda menjadi paket JAR dan unggah menggunakan fitur Artifacts di Konsol Realtime Compute. Setelah paket JAR diunggah dan direferensikan, atur parameter sink.partitioner dalam klausa WITH ke nama kelas lengkap partitioner Anda, misalnya org.mycompany.MyPartitioner.

Kafka, Upsert Kafka, dan katalog Kafka JSON

Kafka adalah platform streaming event yang hanya menambah (append-only) dan tidak mendukung pembaruan atau penghapusan data. Dalam SQL streaming, tabel sink Kafka standar tidak dapat menangani data Change Data Capture (CDC) hulu atau logika penarikan kembali (retraction) dari operator seperti agregat dan join. Jika Anda perlu menulis data yang berisi perubahan atau penarikan kembali, gunakan tabel sink Upsert Kafka.

Untuk menyederhanakan sinkronisasi batch data Change Data Capture (CDC) dari satu atau beberapa tabel database hulu ke Kafka, Anda dapat menggunakan katalog Kafka JSON. Jika data yang disimpan di Kafka dalam format JSON, katalog Kafka JSON memungkinkan Anda melewati langkah mendefinisikan skema dan parameter WITH. Untuk detailnya, lihat Kelola katalog Kafka JSON.

Contoh

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    -- Opsional. Meratakan semua kolom bersarang.
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--Gunakan COALESCE untuk menangani nilai null.

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "Alibaba Cloud",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

Semantik tepat-sekali

• Konfigurasi tingkat isolasi konsumen

  Semua aplikasi yang mengonsumsi data Kafka harus mengatur properti isolation.level:

  • read_committed: Hanya membaca data yang telah dikomit.

  • read_uncommitted (Default): Dapat membaca data yang belum dikomit.

  EXACTLY_ONCE bergantung pada read_committed. Jika tidak, konsumen mungkin melihat data yang belum dikomit, sehingga mengganggu konsistensi.

• Timeout transaksi dan kehilangan data

  Saat memulihkan dari checkpoint, Realtime Compute for Apache Flink hanya mempertimbangkan transaksi yang telah dikomit sebelum checkpoint tersebut dimulai. Jika durasi antara kegagalan pekerjaan dan restart-nya melebihi timeout transaksi Kafka, Kafka secara otomatis membatalkan transaksi yang terbuka, yang dapat menyebabkan kehilangan data.

  • transaction.max.timeout.ms default untuk broker Kafka adalah 15 menit.

  • Secara default, Flink Kafka Sink mengatur parameter transaction.timeout.ms ke 1 jam.

  • Anda harus menaikkan transaction.max.timeout.ms pada broker agar lebih besar dari atau sama dengan pengaturan di Flink.

• Pool produsen dan checkpoint konkuren

  Mode EXACTLY_ONCE menggunakan pool produsen Kafka berukuran tetap. Setiap checkpoint menggunakan satu produsen dari pool ini. Jika jumlah checkpoint konkuren melebihi ukuran pool, pekerjaan gagal.

  Konfigurasikan ukuran pool produsen berdasarkan jumlah maksimum checkpoint konkuren.

• Batasan skala turun paralelisme

  Jika pekerjaan gagal sebelum checkpoint pertama selesai, informasi pool produsen asli hilang saat restart. Oleh karena itu, jangan melakukan skala turun paralelisme pekerjaan sebelum checkpoint pertama selesai. Jika skala turun diperlukan, paralelisme baru tidak boleh kurang dari FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Transaksi memblokir pembacaan

  Dalam mode read_committed, transaksi apa pun yang belum dikomit atau dibatalkan memblokir operasi baca pada seluruh topik.

  Contoh:

  • Transaksi 1 menulis data.

  • Transaksi 2 menulis lebih banyak data dan dikomit.

  • Selama Transaksi 1 tetap terbuka, data dari Transaksi 2 yang telah dikomit tidak terlihat oleh konsumen.

  Hal ini memiliki implikasi berikut:

  • Selama operasi normal, latensi visibilitas data kira-kira sama dengan interval checkpoint.

  • Jika pekerjaan gagal, topik apa pun yang sedang ditulisnya diblokir untuk konsumen sampai pekerjaan dijalankan ulang atau transaksi timeout. Dalam kasus ekstrem, proses timeout transaksi itu sendiri juga dapat memengaruhi operasi baca.

FAQ

• Bagaimana cara mengambil data JSON dari Kafka menggunakan Realtime Compute for Apache Flink?

• Mengapa Window berbasis Event Time tidak menghasilkan output untuk Tabel Sumber ApsaraMQ for Kafka?

• Bagaimana cara menghubungkan ke Kluster Kafka yang diamankan?

• Bagaimana cara menyelesaikan bentrok nama bidang?

• Ruang Kerja dan Namespace