作者：hello簞調_290 | 来源：互联网 | 2023-10-15 15:43
Checklist
- [x] I have included information about relevant versions
- [ ] I have verified that the issue persists when using the
branch of Faust.
Steps to reproduce
I have a
topic with 300,000 messages. Using the code below, with exactly_once enabled, I tested the consumption and the count of the
3 times. Always 300,000 if the process is not interrupted. Then, I tested a fourth time and killed the faust process in the middle of the consumption, then restarted it to finish. The table counter was missing a lot of events. The final count was 299,356.
Expected behavior
I expected not to lose events, if I have
support and the faust process is abruptly terminated and then, restarted. So, in this case I expected to have 300,000 as the final count.
Actual behavior
The final count was wrong.
Full traceback
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| python
import faust
app &＃061; faust.App(
&＃039;count_events&＃039;,
processing_guarantee&＃061;&＃039;exactly_once&＃039;,
broker&＃061;&＃039;kafka://localhost:19092&＃039;,
store&＃061;&＃039;rocksdb://&＃039;,
version&＃061;VERSION
)
# source topic
source &＃061; app.topic(&＃039;example-source-topic&＃039;)
# table
table &＃061; app.Table(&＃039;totals&＃039;, default&＃061;int)
.agent(source)
async def process(stream):
async for event in stream.events():
table[&＃039;total&＃039;] &＃043;&＃061; 1
.page(&＃039;/table/{total}/&＃039;)
.table_route(table&＃061;table, match_info&＃061;&＃039;total&＃039;)
async def get_count(web, request, total):
return web.json({
total: table[&＃039;total&＃039;],
}) |
Versions
- Python 3.6.7
- Faust 1.6.0
- Operating system Ubuntu 18.10
- Kafka version 2.2.0
- RocksDB version (if applicable) 0.7.0
该提问来源于开源项目：robinhood/faust
The source topic in your example is not partitioned? Are you running a different example?
京公网安备 11010802041100号