### Question:
ExactTriangleCount { public static KTable<Long, Long> countTriangles(KGraphStream<Long, Void, Void> graph) { return graph.buildNeighborhood(false) .map(new ExactTriangleCount.ProjectCanonicalEdges()) .flatMap(new ExactTriangleCount.IntersectNeighborhoods()) .mapValues(new ExactTriangleCount.SumAndEmitCounters()) .groupByKey() .reduce(Math::max, Materialized.as(KGraph.generateStoreName())); }  static KTable<Long, Long> countTriangles(KGraphStream<Long, Void, Void> graph);  }### Answer:
@Test public void test() throws Exception { Properties producerConfig = ClientUtils.producerConfig(CLUSTER.bootstrapServers(), LongSerializer.class, LongSerializer.class, new Properties() ); StreamsBuilder builder = new StreamsBuilder(); int numPartitions = 1; KStream<Edge<Long>, Void> edges = StreamUtils.streamFromCollection(builder, producerConfig, "temp-" + UUID.randomUUID(), numPartitions, (short) 1, new KryoSerde<>(), new KryoSerde<>(), getEdges() ); KGraphStream<Long, Void, Void> graph = new EdgeStream<>(edges, GraphSerialized.with(new KryoSerde<>(), new KryoSerde<>(), new KryoSerde<>())); KTable<Long, Long> result = ExactTriangleCount.countTriangles(graph); startStreams(builder, new KryoSerde<>(), new KryoSerde<>()); Thread.sleep(10000); List<String> values = StreamUtils.listFromTable(streams, result).stream() .map(kv -> "(" + kv.key.toString() + "," + kv.value.toString() + ")") .collect(Collectors.toList()); assertEquals( "[(-1,4), (1,2), (2,2), (3,4), (4,1), (5,1), (6,2)]", values.toString() ); streams.close(); }