！！测试环境
-
每秒发送一次带有“A”键的消息的制作人 .
-
Kafka Streams（Java代码）每隔10秒翻滚窗口 .
应该接收每10秒
-
10条消息，并且定期重复发生，如下所示 .
-
必须解决这个问题 .
!!目前的结果
（timestemp，key，每个翻滚窗口的消息数）
（1544079700000L，'A'，10）
（1544079710000L，'A'，10）
（1544079720000L，'A'，10）
(1544079730000L, 'A', 9)
(1544079730000L, 'A', 10)
（1544079740000L，'A'，10）
（1544079750000L，'A'，10）
(1544079760000L, 'A', 9)
(1544079760000L, 'A', 10)
（1544079770000L，'A'，10）
（1544079780000L，'A'，10）
（1544079790000L，'A'，9）
！ TumblingWindowKafkaStream.java（使用Kafka Streams）`
ackage io.github.timothyrenner.kstreamex.tumblingwindow;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import java.util.Properties;
import java.util.Random;
/** Demonstrates tumbling windows.
*
* @author Timothy Renner
*/
public class TumblingWindowKafkaStream {
/** Runs the streams program, writing to the "long-counts-all" topic.
*
* @param args Not used.
*/
public static void main(String[] args) throws Exception {
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG,
"tumbling-window-kafka-streams");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9090,localhost:9091,localhost:9092");
config.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG,
"localhost:2181");
config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
Serdes.ByteArray().getClass().getName());
config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
Serdes.Long().getClass().getName());
KStreamBuilder builder = new KStreamBuilder();
KStream<byte[], byte[]> longs = builder.stream(
Serdes.ByteArray(), Serdes.ByteArray(), "t2");
// The tumbling windows will clear every ten seconds.
KTable<Windowed<byte[]>, Long> longCounts =
longs.groupByKey()
.count(TimeWindows.of(10000L)
.until(10000L),
"t2-counts");
// Write to topics.
longCounts.toStream((k,v) -> k.key())
.to(Serdes.ByteArray(),
Serdes.Long(),
"t2-counts-all");
KafkaStreams streams = new KafkaStreams(builder, config);
streams.start();
} // Close main.
} // Close TumblingWindowKafkaStream.
`