confluentgist

driver.pipeInput(recordFactory.create(
  /* topic */     "input",
  /* key */       "A",
  /* value */     "v1",
  /* timestamp */ 10L
));
// Stream time is now 10L

driver.pipeInput(recordFactory.create("input", "A", "v2", 11L));
// Stream time is now 11L

confluentgist

builder
   .table("users")
   .suppress(Suppressed.untilTimeLimit(
     BufferConfig.maxBytes(myConfig.getUsersBufferSize())
   ))
   ...

confluentgist

events
  .groupByKey()
  .windowedBy(
    TimeWindows.of(Duration.ofMinutes(2).withGrace(Duration.ofMinutes(2))
  )
  .count(Materialized.as("count-metric"))
  .suppress(Suppressed.untilWindowClose(BufferConfig.unbounded()))
  .filter( _ < 4 )
  .toStream()
  .foreach( /* Send that email! */)

confluentgist

events
  .groupByKey()
  .windowedBy(TimeWindows.of(Duration.ofMinutes(2)))
  .count(Materialized.as("count-metric"))
  .filter( _ < 4 )
  .toStream()
  .foreach( /* Send that email! */)

// graph servlet queries "count-metric"

confluentgist

events
  .groupByKey()
  .windowedBy(TimeWindows.of(Duration.ofMinutes(2)))
  .count(Materialized.as("count-metric"))

// graph servlet queries "count-metric"

confluentgist

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *

confluentgist

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *

confluentgist

// imports and license left out for clarity
public class OptimizedStreams {

    public static void main(String[] args) {
        final Properties properties = new Properties();
        
        properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "test-application");
        properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092   ");
        properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION, StreamsConfig.OPTIMIZE);
        properties.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

confluentgist

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *

confluentgist

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *