馬老師kubernetes實(shí)戰(zhàn)與源碼剖析
?@Test
?void helloWorld() {
? ?// kafka config
? ?Properties props = buildConfigProps();
? ?Serde<String> stringSerde = Serdes.String();
? ?props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, stringSerde.getClass().getName());
? ?props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, stringSerde.getClass().getName());
? ?props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_BETA);
? ?props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10000);
標(biāo)簽：
