Hazelcast Jet StreamSourceStage Arayüzü - Timestamp İçindir, Mutlaka Gerekir

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.pipeline.StreamSourceStage;

Kalıtım şöyle

StreamSourceStage

  StreamSourceStageImpl : Tüm metodları bir StreamStageImpl döner

Bu sınıfın bütün metodları bir StreamStage nesnesi döndürür. Aslında StreamSourceStage nesnesinden yani source nesneden StreamStage nesnesine yani stage nesnesine geçmek için aşağıdaki metodlardan birisini mutlaka çağırmak gerekir.

withIngestionTimestamps metodu

İmzası şöyle. 

StreamStage<T> withIngestionTimestamps();

Açıklaması şöyle

Declares that the source will assign the time of ingestion as the event timestamp.

Duvar saatini kullanır. Örnek ver

withNativeTimestamps metodu - Event time processing

İmzası şöyle. 

StreamStage<T> withNativeTimestamps(long allowedLag);

Açıklaması şöyle

Declares that the stream will use the source's native timestamps.

Açıklaması şöyle

The allowedLag parameter is used to specify the maximum amount of delay (in milliseconds) that Jet will tolerate between events that arrive out of order. If the difference between the actual timestamp of an event and its expected timestamp (calculated from its order in the stream) exceeds this allowed lag, Jet will treat it as an out-of-order event and adjust the processing accordingly. The default value of allowedLag is 0, meaning Jet will not tolerate any out-of-order events by default.

Timestamp bilgisi nesne içindedir. Örnek ver

withoutTimestamps metodu

İmzası şöyle. 

StreamStage<T> withoutTimestamps();

Açıklaması şöyle

Declares that the source will not assign any timestamp to the events it emits.

Örnek

Şöyle yaparız

Pipeline cdcPipeline = Pipeline.create();

cdcPipeline
  .readFrom(source)
  .withoutTimestamps()
  .filter(deletedFalse)
  .writeTo(Sinks.map("mapStore", e -> e.key(), e -> e.value()));

jetService.newJobIfAbsent(cdcPipeline);

// now Jet streams data from your Postgres DB to the IMap in the background

withTimestamps metodu - Event time processing

İmzası şöyle.

StreamStage<T> withTimestamps(ToLongFunctionEx<? super T> timestampFn,
  long allowedLag);

Açıklaması şöyle

Declares that the source will extract timestamps from the stream items.

Timestamp bilgisi nesne içindedir. 

Örnek

Şöyle yaparız

.withTimestamps(r -> r.getTimestamp())