Skip to content

Instantly share code, notes, and snippets.

Avatar

Confluent, Inc. confluentgist

View GitHub Profile
View testing-your-suppressions.java
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
View reconfiguring-buffer-size-at-runtime.java
builder
.table("users")
.suppress(Suppressed.untilTimeLimit(
BufferConfig.maxBytes(myConfig.getUsersBufferSize())
))
...
View metrics-app-with-alerts.java
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! */)
View use-case-alerting.java
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"
View suppress.java
events
.groupByKey()
.windowedBy(TimeWindows.of(Duration.ofMinutes(2)))
.count(Materialized.as("count-metric"))
// graph servlet queries "count-metric"
View StreamsDSLExample.java
/*
* 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
*
View StreamsJoinWithRepartitioning.java
/*
* 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
*
View OptimizingStreams.java
// 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());
View ProcessorAPIExample.java
/*
* 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
*
View StreamsDSLAndProcessorExample.java
/*
* 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
*