Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Pertahankan praktik terbaik untuk Layanan Terkelola untuk aplikasi Apache Flink
Bagian ini berisi informasi dan rekomendasi untuk mengembangkan Layanan Terkelola yang stabil dan berkinerja untuk aplikasi Apache Flink.
**Topics**
+ [Minimalkan ukuran Uber JAR](#minimize-uber-JAR)
+ [Toleransi kesalahan: titik pemeriksaan dan titik simpan](#how-dev-bp-checkpoint)
+ [Versi konektor yang tidak didukung](#how-dev-bp-connectors)
+ [Performa dan paralelisme](#how-dev-bp-performance)
+ [Pengaturan paralelisme per operator](#how-dev-bp-parallelism)
+ [Pencatatan log](#how-dev-bp-logging)
+ [Pengkodean](#how-dev-bp-code)
+ [Mengelola kredensyal](#how-dev-bp-managing-credentials)
+ [Membaca dari sumber dengan sedikit pecahan/partisi](#troubleshooting-few-shards-partitions)
+ [Interval refresh notebook Studio](#notebook-refresh-rate)
+ [Performa optimum notebook Studio](#notebook-refresh-rate)
+ [Bagaimana strategi watermark dan pecahan idle memengaruhi jendela waktu](#notebook-watermarking)
+ [Tetapkan UUID untuk semua operator](#best-practices-setting-operator-ids)
+ [Tambahkan ServiceResourceTransformer ke plugin Maven shade](#best-practices-service-resource-transformer)
## Minimalkan ukuran Uber JAR
Java/Scala application must be packaged in an uber (super/fat) JAR dan sertakan semua dependensi tambahan yang diperlukan yang belum disediakan oleh runtime. Namun, ukuran Uber JAR mempengaruhi waktu mulai dan restart aplikasi dan dapat menyebabkan JAR melebihi batas 512 MB.
Untuk mengoptimalkan waktu penerapan, Uber JAR Anda **tidak** boleh menyertakan yang berikut:
+ **Setiap dependensi yang disediakan oleh runtime seperti yang** diilustrasikan dalam contoh berikut. Mereka harus memiliki `provided` ruang lingkup dalam file POM atau `compileOnly` dalam konfigurasi Gradle Anda.
+ **Setiap dependensi yang digunakan untuk pengujian saja**, misalnya JUnit atau Mockito. Mereka harus memiliki `test` ruang lingkup dalam file POM atau `testImplementation` dalam konfigurasi Gradle Anda.
+ **Dependensi apa pun yang sebenarnya tidak digunakan oleh aplikasi Anda**.
+ **Setiap data statis atau metadata yang diperlukan oleh aplikasi Anda.** Data statis harus dimuat oleh aplikasi saat runtime, misalnya dari datastore atau dari Amazon S3.
+ Lihat [file contoh POM](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/blob/main/java/GettingStarted/pom.xml) ini untuk detail tentang pengaturan konfigurasi sebelumnya.
**Dependensi yang disediakan**
Managed Service for Apache Flink runtime menyediakan sejumlah dependensi. Dependensi ini tidak boleh dimasukkan dalam JAR gemuk dan harus memiliki `provided` ruang lingkup dalam file POM atau secara eksplisit dikecualikan dalam konfigurasi. `maven-shade-plugin` Salah satu dependensi ini yang termasuk dalam JAR gemuk diabaikan saat runtime, tetapi meningkatkan ukuran JAR yang menambahkan overhead selama penerapan.
Dependensi disediakan oleh runtime, dalam runtime versi 1.18, 1.19, dan 1.20:
+ `org.apache.flink:flink-core`
+ `org.apache.flink:flink-java`
+ `org.apache.flink:flink-streaming-java`
+ `org.apache.flink:flink-scala_2.12`
+ `org.apache.flink:flink-table-runtime`
+ `org.apache.flink:flink-table-planner-loader`
+ `org.apache.flink:flink-json`
+ `org.apache.flink:flink-connector-base`
+ `org.apache.flink:flink-connector-files`
+ `org.apache.flink:flink-clients`
+ `org.apache.flink:flink-runtime-web`
+ `org.apache.flink:flink-metrics-code`
+ `org.apache.flink:flink-table-api-java`
+ `org.apache.flink:flink-table-api-bridge-base`
+ `org.apache.flink:flink-table-api-java-bridge`
+ `org.apache.logging.log4j:log4j-slf4j-impl`
+ `org.apache.logging.log4j:log4j-api`
+ `org.apache.logging.log4j:log4j-core`
+ `org.apache.logging.log4j:log4j-1.2-api`
Selain itu, runtime menyediakan library yang digunakan untuk mengambil properti runtime aplikasi di Managed Service for Apache Flink. `com.amazonaws:aws-kinesisanalytics-runtime:1.2.0`
Semua dependensi yang disediakan oleh runtime harus menggunakan rekomendasi berikut untuk **tidak** memasukkannya ke dalam Uber JAR:
+ Di Maven (`pom.xml`) dan SBT (`build.sbt`), gunakan ruang lingkup. `provided`
+ Di Gradle (`build.gradle`), gunakan `compileOnly` konfigurasi.
Ketergantungan apa pun yang disediakan secara tidak sengaja disertakan dalam Uber JAR akan diabaikan saat runtime karena pemuatan kelas induk-pertama Apache Flink. Untuk informasi lebih lanjut, lihat [parent-first-patterns](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#classloader-parent-first-patterns-default)di dokumentasi Apache Flink.
**Konektor**
Sebagian besar konektor, kecuali FileSystem konektor, yang tidak termasuk dalam runtime harus disertakan dalam file POM dengan cakupan default ()`compile`.
**Rekomendasi lainnya**
Sebagai aturan, Apache Flink uber JAR yang disediakan untuk Managed Service for Apache Flink harus berisi kode minimum yang diperlukan untuk menjalankan aplikasi. Menyertakan dependensi yang menyertakan kelas sumber, kumpulan data pengujian, atau status bootstrap tidak boleh disertakan dalam toples ini. Jika sumber daya statis perlu ditarik saat runtime, pisahkan masalah ini menjadi sumber daya seperti Amazon S3. Contohnya termasuk bootstraps status atau model inferensi.
Luangkan waktu untuk mempertimbangkan pohon ketergantungan mendalam Anda dan hapus dependensi non-runtime.
Meskipun Managed Service untuk Apache Flink mendukung ukuran jar 512MB, ini harus dilihat sebagai pengecualian aturan. Apache Flink saat ini mendukung ukuran jar \$1 104MB melalui konfigurasi defaultnya, dan itu harus menjadi ukuran target maksimum dari toples yang dibutuhkan.
## Toleransi kesalahan: titik pemeriksaan dan titik simpan
Gunakan pos pemeriksaan dan savepoint untuk menerapkan toleransi kesalahan dalam Layanan Terkelola untuk aplikasi Apache Flink Anda. Ingat hal berikut saat mengembangkan dan memelihara aplikasi Anda:
+ Kami menyarankan agar Anda tetap mengaktifkan checkpointing untuk aplikasi Anda. Checkpointing memberikan toleransi kesalahan untuk aplikasi Anda selama pemeliharaan terjadwal, dan juga untuk kegagalan tak terduga karena masalah layanan, kegagalan ketergantungan aplikasi, dan masalah lainnya. Untuk informasi tentang pemeliharaan terjadwal, lihat [Mengelola tugas pemeliharaan untuk Managed Service untuk Apache Flink](maintenance.md).
+ Set [ApplicationSnapshotConfiguration:: SnapshotsEnabled](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ApplicationSnapshotConfiguration.html) ke `false` selama pengembangan aplikasi atau pemecahan masalah. Snapshot dibuat selama setiap aplikasi berhenti, yang dapat menyebabkan masalah jika aplikasi dalam keadaan tidak sehat atau tidak berkinerja. Atur `SnapshotsEnabled` ke `true` setelah aplikasi dalam produksi dan stabil.
**catatan**
Kami menyarankan Anda mengatur aplikasi Anda untuk membuat snapshot beberapa kali sehari untuk memulai ulang dengan benar dengan data status yang benar. Frekuensi yang benar untuk snapshot Anda bergantung pada logika bisnis aplikasi Anda. Mengambil snapshot yang sering memungkinkan Anda memulihkan data yang lebih baru, tetapi meningkatkan biaya dan membutuhkan lebih banyak sumber daya sistem.
Untuk informasi tentang pemantauan waktu henti aplikasi, lihat [Metrik dan dimensi dalam Layanan Terkelola untuk Apache Flink](metrics-dimensions.md).
Untuk informasi selengkapnya tentang penerapan toleransi kegagalan, lihat [Menerapkan toleransi kesalahan](how-fault.md).
## Versi konektor yang tidak didukung
Dari Apache Flink versi 1.15 atau yang lebih baru, Managed Service for Apache Flink secara otomatis mencegah aplikasi memulai atau memperbarui jika mereka menggunakan versi konektor Kinesis yang tidak didukung yang dibundel ke dalam aplikasi. JARs Saat memutakhirkan ke Managed Service untuk Apache Flink versi 1.15 atau yang lebih baru, pastikan Anda menggunakan konektor Kinesis terbaru. Ini adalah versi apa pun yang sama dengan atau lebih baru dari versi 1.15.2. Semua versi lain tidak didukung oleh Managed Service untuk Apache Flink karena mereka dapat menyebabkan masalah konsistensi atau kegagalan dengan fitur **Stop with Savepoint**, mencegah operasi berhenti/pembaruan bersih. Untuk mempelajari lebih lanjut tentang kompatibilitas konektor di Amazon Managed Service untuk versi Apache Flink, lihat konektor [Apache](https://docs.aws.amazon.com/managed-flink/latest/java/how-flink-connectors.html) Flink.
## Performa dan paralelisme
Aplikasi Anda dapat diskalakan untuk memenuhi tingkat throughput apa pun dengan menyetel paralelisme aplikasi Anda, dan menghindari perangkap performa. Ingat hal berikut saat mengembangkan dan memelihara aplikasi Anda:
+ Verifikasi bahwa semua sumber aplikasi dan sink Anda ditetapkan dengan cukup dan tidak dibatasi. Jika sumber dan wastafel adalah AWS layanan lain, pantau layanan tersebut menggunakan [CloudWatch](https://docs.aws.amazon.com/cloudwatch/?id=docs_gateway).
+ Untuk aplikasi dengan paralelisme yang sangat tinggi, periksa apakah tingkat paralelisme yang tinggi diterapkan pada semua operator dalam aplikasi. Secara default, Apache Flink menerapkan paralelisme aplikasi yang sama untuk semua operator dalam grafik aplikasi. Ini dapat menyebabkan masalah penyediaan pada sumber atau sink, atau pun hambatan dalam pemrosesan data operator. Anda dapat mengubah paralelisme setiap operator dalam kode dengan [setParallelism](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/parallel.html).
+ Pahami arti pengaturan paralelisme untuk operatori dalam aplikasi Anda. Jika Anda mengubah paralelisme untuk operator, Anda mungkin tidak dapat memulihkan aplikasi dari snapshot yang dibuat ketika operator memiliki paralelisme yang tidak kompatibel dengan pengaturan saat ini. Untuk informasi selengkapnya tentang pengaturan paralelisme operator, lihat [ Mengatur paralelisme maksimum untuk operator secara eksplisit](https://nightlies.apache.org/flink/flink-docs-release-1.15/ops/production_ready.html#set-maximum-parallelism-for-operators-explicitly).
Untuk informasi selengkapnya tentang penerapan penskalaan, lihat [Menerapkan penskalaan aplikasi](how-scaling.md).
## Pengaturan paralelisme per operator
Secara default, semua operator memiliki paralelisme yang ditetapkan pada tingkat aplikasi. Anda dapat mengganti paralelisme dari satu operator menggunakan API. DataStream `.setParallelism(x)` Anda dapat mengatur paralelisme operator ke paralelisme apa pun yang sama atau lebih rendah dari paralelisme aplikasi.
Jika memungkinkan, tentukan paralelisme operator sebagai fungsi dari paralelisme aplikasi. Dengan cara ini, paralelisme operator akan bervariasi dengan paralelisme aplikasi. Jika Anda menggunakan penskalaan otomatis, misalnya, semua operator akan memvariasikan paralelisme mereka dalam proporsi yang sama:
```
int appParallelism = env.getParallelism();
...
...ops.setParalleism(appParallelism/2);
```
Dalam beberapa kasus, Anda mungkin ingin mengatur paralelisme operator ke konstanta. Misalnya, mengatur paralelisme sumber Aliran Kinesis ke jumlah pecahan. Dalam kasus ini, pertimbangkan untuk meneruskan paralelisme operator sebagai parameter konfigurasi aplikasi untuk mengubahnya tanpa mengubah kode, misalnya untuk mengubah aliran sumber.
## Pencatatan log
Anda dapat memantau kinerja dan kondisi kesalahan aplikasi Anda menggunakan CloudWatch Log. Ingat hal berikut saat mengonfigurasi pencatatan untuk aplikasi Anda:
+ Aktifkan CloudWatch pencatatan untuk aplikasi sehingga masalah runtime apa pun dapat di-debug.
+ Jangan buat entri log untuk setiap catatan yang diproses dalam aplikasi. Hal ini menyebabkan kemacetan parah selama pemrosesan dan dapat menyebabkan tekanan balik dalam pemrosesan data.
+ Buat CloudWatch alarm untuk memberi tahu Anda ketika aplikasi Anda tidak berjalan dengan benar. Untuk informasi selengkapnya, lihat [Gunakan CloudWatch Alarm dengan Amazon Managed Service untuk Apache Flink](monitoring-metrics-alarms.md)
Untuk informasi selengkapnya tentang penerapan pencatatan, lihat [](monitoring-overview.md).
## Pengkodean
Anda dapat membuat aplikasi Anda berfungsi dan stabil menggunakan praktik pemrograman yang direkomendasikan. Ingat hal berikut saat menulis kode aplikasi:
+ Jangan gunakan `system.exit()` dalam kode aplikasi Anda, baik dalam metode `main` aplikasi Anda atau dalam fungsi yang ditetapkan pengguna. Jika Anda ingin menonaktifkan aplikasi Anda dari dalam kode, lempar pengecualian yang berasal dari `Exception` atau `RuntimeException`, yang berisi pesan tentang apa yang salah dengan aplikasi.
Catat hal berikut tentang bagaimana layanan menangani pengecualian ini:
+ Jika pengecualian dilemparkan dari metode `main` aplikasi Anda, layanan akan membungkusnya dalam `ProgramInvocationException` saat transisi aplikasi ke status `RUNNING`, dan manajer tugas akan gagal mengirimkan tugas.
+ Jika pengecualian dilemparkan dari fungsi yang ditetapkan pengguna, manajer tugas akan gagal tugas dan memulai ulang, serta detail pengecualian akan ditulis ke log pengecualian.
+ Pertimbangkan bayangan file JAR aplikasi Anda dan dependensi yang disertakan. Bayangan direkomendasikan ketika ada potensi konflik dalam nama paket antara aplikasi Anda dan runtime Apache Flink. Jika terjadi konflik, log aplikasi Anda mungkin berisi pengecualian tipe `java.util.concurrent.ExecutionException`. Untuk informasi selengkapnya tentang bayangan file JAR aplikasi Anda, lihat [Plugin Apache Maven Shade](https://maven.apache.org/plugins/maven-shade-plugin/).
## Mengelola kredensyal
Anda tidak boleh memanggang kredensi jangka panjang apa pun ke dalam aplikasi produksi (atau lainnya). Kredensi jangka panjang kemungkinan diperiksa ke dalam sistem kontrol versi dan dapat dengan mudah hilang. Sebagai gantinya, Anda dapat mengaitkan peran ke aplikasi Managed Service for Apache Flink dan memberikan izin ke peran tersebut. Aplikasi Flink yang sedang berjalan kemudian dapat memilih kredensyal sementara dengan izin masing-masing dari lingkungan. [Dalam hal otentikasi diperlukan untuk layanan yang tidak terintegrasi secara native dengan IAM, misalnya, database yang memerlukan nama pengguna dan kata sandi untuk otentikasi, Anda harus mempertimbangkan untuk menyimpan rahasia di Secrets Manager.AWS](https://aws.amazon.com/secrets-manager/)
Banyak layanan AWS asli mendukung otentikasi:
+ [Kinesis Data ProcessTaxiStream Streams — .java](hhttps://github.com/aws-samples/amazon-kinesis-data-analytics-taxi-consumer/blob/master/src/main/java/com/amazonaws/samples/kaja/taxi/consumer/ProcessTaxiStream.java#L90)
+ Amazon MSK — [https://github.com/aws/aws-msk-iam-authusing-the-amazon-msk/\$1](https://github.com/aws/aws-msk-iam-auth/#using-the-amazon-msk-library-for-iam-authentication) - library-for-iam-authentication
+ [Amazon Elasticsearch Service — .java AmazonElasticsearchSink](https://github.com/aws-samples/amazon-kinesis-data-analytics-taxi-consumer/blob/master/src/main/java/com/amazonaws/samples/kaja/taxi/consumer/operators/AmazonElasticsearchSink.java)
+ Amazon S3 - bekerja di luar kotak pada Layanan Terkelola untuk Apache Flink
## Membaca dari sumber dengan sedikit pecahan/partisi
Saat membaca dari Apache Kafka atau Aliran Data Kinesis, mungkin ada ketidakcocokan antara paralelisme aliran (jumlah partisi untuk Kafka dan jumlah pecahan untuk Kinesis) dan paralelisme aplikasi. Dengan desain yang naif, paralelisme aplikasi tidak dapat berskala melampaui paralelisme aliran: Setiap subtugas operator sumber hanya dapat membaca dari 1 atau lebih piringan/partisi. Itu berarti untuk aliran dengan hanya 2 pecahan dan aplikasi dengan paralelisme 8, bahwa hanya dua subtugas yang benar-benar memakan dari aliran dan 6 subtugas tetap menganggur. Ini secara substansional dapat membatasi throughput aplikasi, khususnya jika deserialisasi mahal dan dilakukan oleh sumber (yang merupakan default).
Untuk mengurangi efek ini, Anda dapat menskalakan aliran. Tapi itu mungkin tidak selalu diinginkan atau mungkin. Atau, Anda dapat merestrukturisasi sumber sehingga tidak melakukan serialisasi apa pun dan hanya meneruskan. `byte[]` Anda kemudian dapat [menyeimbangkan kembali](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/overview/) data untuk mendistribusikannya secara merata di semua tugas dan kemudian deserialisasi data di sana. Dengan cara ini, Anda dapat memanfaatkan semua subtugas untuk deserialisasi dan operasi yang berpotensi mahal ini tidak lagi terikat oleh shards/partitions jumlah aliran.
## Interval refresh notebook Studio
Jika Anda mengubah interval refresh hasil paragraf, atur ke nilai yang setidaknya 1000 milidetik.
## Performa optimum notebook Studio
Kami menguji dengan pernyataan berikut dan mendapatkan kinerja optimal ketika `events-per-second` dikalikan dengan di bawah `number-of-keys` 25.000.000. Ini adalah untuk `events-per-second` di bawah 150.000.
```
SELECT key, sum(value) FROM key-values GROUP BY key
```
## Bagaimana strategi watermark dan pecahan idle memengaruhi jendela waktu
Saat membaca peristiwa dari Apache Kafka dan Kinesis Data Streams, sumber dapat mengatur waktu acara berdasarkan atribut aliran. Dalam kasus Kinesis, waktu acara sama dengan perkiraan waktu kedatangan peristiwa. Tetapi pengaturan waktu acara di sumber untuk acara tidak cukup bagi aplikasi Flink untuk menggunakan waktu acara. Sumber juga harus menghasilkan tanda air yang menyebarkan informasi tentang waktu acara dari sumber ke semua operator lain. [Dokumentasi Flink](https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/time/) memiliki gambaran yang baik tentang cara kerja proses itu.
Secara default, stempel waktu peristiwa yang dibaca dari Kinesis diatur ke perkiraan waktu kedatangan yang ditentukan oleh Kinesis. Prasyarat tambahan untuk waktu acara untuk bekerja dalam aplikasi adalah strategi watermark.
```
WatermarkStrategy s = WatermarkStrategy
.forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(...));
```
Strategi watermark kemudian diterapkan ke a `DataStream` with the `assignTimestampsAndWatermarks` method. Ada beberapa strategi bawaan yang berguna:
+ `forMonotonousTimestamps()`hanya akan menggunakan waktu acara (perkiraan waktu kedatangan) dan secara berkala memancarkan nilai maksimum sebagai tanda air (untuk setiap subtugas tertentu)
+ `forBoundedOutOfOrderness(Duration.ofSeconds(...))`mirip dengan strategi sebelumnya, tetapi akan menggunakan waktu acara - durasi untuk pembuatan tanda air.
Dari [dokumentasi Flink](https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/time/):
*Setiap subtugas paralel dari fungsi sumber biasanya menghasilkan tanda airnya secara independen. Tanda air ini menentukan waktu acara pada sumber paralel tertentu.*
*Saat tanda air mengalir melalui program streaming, mereka memajukan waktu acara di operator tempat mereka tiba. Setiap kali operator memajukan waktu acaranya, ia menghasilkan tanda air baru di hilir untuk operator penggantinya.*
*Beberapa operator menggunakan beberapa aliran input; serikat pekerja, misalnya, atau operator yang mengikuti fungsi KeyBy (...) atau partisi (...). Waktu kejadian operator saat ini adalah minimum waktu acara aliran inputnya. Karena aliran inputnya memperbarui waktu acara mereka, begitu juga operator.*
Itu berarti, jika subtugas sumber mengkonsumsi dari pecahan siaga, operator hilir tidak menerima tanda air baru dari subtugas itu dan karenanya memproses stall untuk semua operator hilir yang menggunakan jendela waktu. Untuk menghindari hal ini, pelanggan dapat menambahkan `withIdleness` opsi ke strategi tanda air. Dengan opsi itu, operator mengecualikan tanda air dari subtugas upstream idle saat menghitung waktu acara operator. Oleh karena itu, subtugas idle tidak lagi memblokir kemajuan waktu acara di operator hilir.
Bergantung pada pemecah pecahan yang Anda gunakan, beberapa pekerja mungkin tidak diberi pecahan Kinesis apa pun. Dalam hal ini, pekerja ini akan memanifestasikan perilaku sumber menganggur bahkan jika semua pecahan Kinesis terus mengirimkan data peristiwa. Anda dapat mengurangi risiko ini dengan menggunakan `uniformShardAssigner` dengan operator sumber. Ini memastikan bahwa semua subtugas sumber memiliki pecahan untuk diproses selama jumlah pekerja kurang atau sama dengan jumlah pecahan aktif.
Namun, opsi kemalasan dengan strategi tanda air bawaan tidak akan memajukan waktu acara jika tidak ada subtugas yang membaca acara apa pun, yaitu tidak ada acara dalam aliran. Ini menjadi sangat terlihat untuk kasus uji di mana serangkaian peristiwa terbatas dibaca dari aliran. Karena waktu acara tidak berlanjut setelah acara terakhir dibaca, jendela terakhir (berisi acara terakhir) tidak akan ditutup.
### Ringkasan
+ `withIdleness`Pengaturan tidak akan menghasilkan tanda air baru jika pecahan tidak digunakan. Ini akan mengecualikan tanda air terakhir yang dikirim oleh subtugas idle dari perhitungan tanda air min di operator hilir.
+ Dengan strategi watermark bawaan, jendela terbuka terakhir tidak akan ditutup (kecuali peristiwa baru yang memajukan tanda air akan dikirim, tetapi itu menciptakan jendela baru yang kemudian tetap terbuka).
+ Bahkan ketika waktu diatur oleh aliran Kinesis, peristiwa kedatangan terlambat masih dapat terjadi jika satu pecahan dikonsumsi lebih cepat daripada yang lain (misalnya selama inisialisasi aplikasi atau saat menggunakan `TRIM_HORIZON` di mana semua pecahan yang ada dikonsumsi secara paralel mengabaikan hubungannya). parent/child
+ `withIdleness`Pengaturan strategi tanda air tampaknya mengganggu pengaturan khusus sumber Kinesis untuk pecahan siaga. `(ConsumerConfigConstants.SHARD_IDLE_INTERVAL_MILLIS`
### Contoh
Aplikasi berikut membaca dari aliran dan membuat jendela sesi berdasarkan waktu acara.
```
Properties consumerConfig = new Properties();
consumerConfig.put(AWSConfigConstants.AWS_REGION, "eu-west-1");
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "TRIM_HORIZON");
FlinkKinesisConsumer consumer = new FlinkKinesisConsumer<>("...", new SimpleStringSchema(), consumerConfig);
WatermarkStrategy s = WatermarkStrategy
.forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(15));
env.addSource(consumer)
.assignTimestampsAndWatermarks(s)
.map(new MapFunction() {
@Override
public Long map(String s) throws Exception {
return Long.parseLong(s);
}
})
.keyBy(l -> 0l)
.window(EventTimeSessionWindows.withGap(Time.seconds(10)))
.process(new ProcessWindowFunction() {
@Override
public void process(Long aLong, ProcessWindowFunction.Context context, Iterableiterable, Collector