Working with Time-Based Processing

1. Understanding Time Semantics

Time Type Description Detail
Event Time When event occurred In payload/ts
Processing Time When app processes Wall clock
Ingestion Time When broker appends LogAppendTime

Example: Time concept

event -> produce -> broker append (ingestion) -> consume (processing)

2. Understanding Event Time

Aspect Description Detail
Source Record timestamp or field Producer set
Correctness Deterministic results Reprocessing safe
Used By Windowing default Streams

Example: Default extractor

// FailOnInvalidTimestamp uses record's event timestamp
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
  FailOnInvalidTimestamp.class);

3. Understanding Processing Time

Aspect Description Detail
Source System wall clock Non-deterministic
Pros Simple, no skew handling Low latency
Cons Reprocessing differs Inconsistent

Example: Wall-clock extractor

props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
  WallclockTimestampExtractor.class);

4. Understanding Ingestion Time

Aspect Description Detail
Set By Broker on append LogAppendTime
Config message.timestamp.type Topic-level
Use Middle ground Approximate event

Example: Set ingestion time

kafka-configs.sh --bootstrap-server localhost:9092 --alter \
  --entity-type topics --entity-name t \
  --add-config message.timestamp.type=LogAppendTime

5. Configuring Timestamp Extractors

Extractor Description Detail
FailOnInvalidTimestamp Default, throws on negative Strict
LogAndSkipOnInvalid Skips bad timestamps Lenient
UsePartitionTime Fallback to stream time Recovery

Example: Per-stream extractor

builder.stream("in",
  Consumed.with(Serdes.String(), Serdes.String())
    .withTimestampExtractor(new OrderTimestampExtractor()));

6. Implementing Custom Timestampers

Method Description Detail
extract() Return event time millis From value
partitionTime Previous max as fallback Param
Negative Signals invalid Skipped

Example: Custom extractor

public long extract(ConsumerRecord<Object,Object> r, long partitionTime){
  Order o = (Order) r.value();
  return o != null ? o.getCreatedAt() : partitionTime;
}

7. Using Tumbling Windows

Aspect Description Detail
Shape Fixed, non-overlapping Size = advance
Each Record Exactly one window No overlap
Use Periodic aggregates Per-minute counts

Example: Tumbling window

TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1));

8. Using Hopping Windows

Aspect Description Detail
Shape Fixed, overlapping Size > advance
Each Record Multiple windows Overlap
Use Smoothed moving stats 5m every 1m

Example: Hopping window

TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
  .advanceBy(Duration.ofMinutes(1));

9. Using Sliding Windows

Aspect Description Detail
Shape Window per record pair Difference-based
timeDifference Max gap between records Inclusive
Use Pairwise aggregation Efficient

Example: Sliding window

SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10));

10. Using Session Windows

Aspect Description Detail
Shape Dynamic, activity-based No fixed size
Inactivity Gap Closes after silence Merge if close
Use User session analytics Clickstream

Example: Session window

SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5));

11. Configuring Grace Period

Aspect Description Detail
Grace Wait for late records Before close
withGrace Replaces deprecated until() Explicit
After Grace Late records dropped Final result

Example: Window with grace

TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5),
  Duration.ofMinutes(1));

12. Handling Out-of-Order Events

Strategy Description Detail
Stream Time Max observed event time Monotonic
Grace Period Accept late within bound Tolerance
Suppress Emit only final per window No early

Example: Suppress until close

.suppress(Suppressed.untilWindowCloses(
  Suppressed.BufferConfig.unbounded()));

Related Cheat Sheets

Kafka

Configuring Producer Settings

Kafka

Implementing Multi-Datacenter Replication

Kafka

Using Schema Registry

Kafka

Using MirrorMaker for Replication