Sendingadeletion/tombstonemessagetoatopicraisesanerror

作者：mobiledu2502884357 | 来源：互联网 | 2023-08-24 18:14

IamtryingtobuildaTablethatmaintainsalistofavailableitemsbyID.Inorderto

I am trying to build a Table that maintains a list of available items by ID. In order to keep the table up-to-date, I need to send deletion messagse (i.e. same key as a previous message, but

1	value&＃061;None

) to a Faust topic. However, this fails. Am I using the API incorrectly or is this an issue with the default serializer?

Checklist

[x] I have included information about relevant versions

[ ] I have verified that the issue persists when using the
1
master
branch of Faust.

Steps to reproduce

Send a message with

1	value&＃061;None

to a topic.

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

py

import faust

app &＃061; faust.App(&＃039;test&＃039;, broker&＃061;&＃039;kafka://localhost:9094&＃039;, store&＃061;&＃039;memory://&＃039;)

class Item(faust.Record):

id: str

val: str

item_topic &＃061; app.topic(&＃039;items&＃039;, value_type&＃061;Item)

items &＃061; app.Table(&＃039;items&＃039;)

.agent(item_topic)

async def build_item_index(stream: faust.Stream):

async for item in stream:

if item:

print(f&＃039;Adding {item.id} to index&＃039;)

items[item.id] &＃061; item

else:

print(f&＃039;Removing {item.id} from index&＃039;)

del items[item.id]

.task

async def on_started():

await item_topic.send(key&＃061;&＃039;1234&＃039;, value&＃061;Item(1, &＃039;a&＃039;))

await item_topic.send(key&＃061;&＃039;1234&＃039;, value&＃061;None) # Causing the error

Expected behavior

Agents that process

1	item_topic

receive an event that is

None

Actual behavior

Agents never receive the deletion event.

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

pytb

[2018-08-21 08:45:09,790: INFO]: [^Worker]: Ready

[2018-08-21 08:45:09,921: WARNING]: Adding 1 to index

[2018-08-21 08:45:09,994: ERROR]: [^---Agent*: api.test.build_item_index]: Crashed reason&＃061;ValueDecodeError(&＃039;type object argument after ** must be a mapping, not NoneType&＃039;,)

Traceback (most recent call last):

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 99, in loads_value

return cast(V, self._prepare_payload(typ, payload))

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 122, in _prepare_payload

return model.from_data(value, preferred_type&＃061;model)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/models/record.py", line 280, in from_data

return (self_cls or cls)(**data, __strict__&＃061;False)

TypeError: type object argument after ** must be a mapping, not NoneType

The above exception was the direct cause of the following exception:

Traceback (most recent call last):

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/agents/agent.py", line 527, in _execute_task

await coro

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/api/test.py", line 14, in build_item_index

async for item in stream:

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/streams.py", line 747, in __aiter__

channel_value &＃061; await chan_slow_get()

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/channels.py", line 301, in __anext__

return await self.queue.get()

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/mode/utils/queues.py", line 125, in get

return await super().get()

File "/usr/lib64/python3.6/asyncio/queues.py", line 167, in get

yield from getter

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/transport/conductor.py", line 80, in on_message

event &＃061; await chan.decode(message, propagate&＃061;True)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/topics.py", line 161, in decode

value_type, message.value, serializer&＃061;value_serializer)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 104, in loads_value

sys.exc_info()[2]) from exc

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 99, in loads_value

return cast(V, self._prepare_payload(typ, payload))

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 122, in _prepare_payload

return model.from_data(value, preferred_type&＃061;model)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/models/record.py", line 280, in from_data

return (self_cls or cls)(**data, __strict__&＃061;False)

faust.exceptions.ValueDecodeError: type object argument after ** must be a mapping, not NoneType

[2018-08-21 08:45:09,997: ERROR]: [^---Agent*: api.test.build_item_index]: Crashed reason&＃061;ValueDecodeError(&＃039;type object argument after ** must be a mapping, not NoneType&＃039;,)

Traceback (most recent call last):

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 99, in loads_value

return cast(V, self._prepare_payload(typ, payload))

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 122, in _prepare_payload

return model.from_data(value, preferred_type&＃061;model)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/models/record.py", line 280, in from_data

return (self_cls or cls)(**data, __strict__&＃061;False)

TypeError: type object argument after ** must be a mapping, not NoneType

The above exception was the direct cause of the following exception:

Traceback (most recent call last):

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/mode/services.py", line 688, in _execute_task

await task

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/agents/agent.py", line 527, in _execute_task

await coro

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/api/test.py", line 14, in build_item_index

async for item in stream:

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/streams.py", line 747, in __aiter__

channel_value &＃061; await chan_slow_get()

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/channels.py", line 301, in __anext__

return await self.queue.get()

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/mode/utils/queues.py", line 125, in get

return await super().get()

File "/usr/lib64/python3.6/asyncio/queues.py", line 167, in get

yield from getter

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/transport/conductor.py", line 80, in on_message

event &＃061; await chan.decode(message, propagate&＃061;True)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/topics.py", line 161, in decode

value_type, message.value, serializer&＃061;value_serializer)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 104, in loads_value

sys.exc_info()[2]) from exc

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 99, in loads_value

return cast(V, self._prepare_payload(typ, payload))

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/serializers/registry.py", line 122, in _prepare_payload

return model.from_data(value, preferred_type&＃061;model)

File "/home/michael/arbeit/digitalernachschub/ameto/ameto/api/venv/lib64/python3.6/site-packages/faust/models/record.py", line 280, in from_data

return (self_cls or cls)(**data, __strict__&＃061;False)

faust.exceptions.ValueDecodeError: type object argument after ** must be a mapping, not NoneType

Versions

Python version: 3.6.5

Faust version: 1.0.30

Operating system: Linux

Kafka version: 1.0.1

RocksDB version (if applicable)

该提问来源于开源项目：robinhood/faust

Thank you!

推荐阅读

int
使用FFMpeg转码(精华)

使用ffmpeg进行视频格式转换的简单例子2006-12-1623:12主要参考FFMPEG里面的apiexample.c以及output_example.c编写intmain(in ... [详细]

蜡笔小新 2024-09-26 10:58:56
import
关于“如何在GAE上托管静态站点”遇到的问题求助！！

http://segmentfault.com/a/1190000000440293?page=1#c-1190000000440293-1050000000449039这是博主原帖 ... [详细]

蜡笔小新 2024-09-28 11:57:01
sum
Spring Cloud笔记Spring Cloud Stream消息驱动（十五）

1.消息驱动概述1.SpringCloudStream是什么SpringCloudStream是一个构建消息驱动微服务的框架。应用程序通过Inpust和Outputs与Spri ... [详细]

蜡笔小新 2024-09-27 14:07:59
int
Java反序列化漏洞（ysoserial工具使用、shiro反序列化利用）

Java反序列化机制Java通过writeObject序列化将对象保存为二进制数据流，通过readObject反序列化将序列化后的二进制重新反序列化为Java对象& ... [详细]

蜡笔小新 2024-09-24 18:14:56
int
什么是大数据lambda架构

一、什么是Lambda架构Lambda架构由Storm的作者[NathanMarz]提出，根据维基百科的定义，Lambda架构的设计是为了在处理大规模数 ... [详细]

蜡笔小新 2023-10-17 16:06:09
import
java中打开文件显示_在默认文件资源管理器中打开文件,并使用JavaFX或普通Java突出显示它...

我想做标题所说的.部分解决方案例如,在Windows中,您可以使用以下代码在默认资源管理器中打开文件并突出显示它.(虽然它需要修改包含空格的文件)：***Openst ... [详细]

蜡笔小新 2024-09-30 08:33:14
web
Linux下安装grafana并且添加influxdb监控的方法

这篇文章主要介绍了Linux下安装grafana并且添加influxdb监控的方法，本文给大家介绍的非常详细，具有一定的参考借鉴价值, ... [详细]

蜡笔小新 2024-09-29 14:04:22
callback
怎么用php登录微博(2023年最新整理)

导读：今天编程笔记来给各位分享关于怎么用php登录微博的相关内容，如果能碰巧解决你现在面临的问题，别忘了关注本站，现在开始吧！本文目录一览： ... [详细]

蜡笔小新 2024-09-28 23:46:59
spring
SpringCloud+SpringBoot 项目搭建（一） Eureka

SpringCloud+SpringBoot项目搭建（一）Eureka-SpringCloud+SpringBoot项目搭建（一）Eureka基础SpringBoot+Eur ... [详细]

蜡笔小新 2024-09-27 14:57:55
int
怎么把mysql中的表以表格的形式输出？

代码如下：（把数据以表格形式输出本人小白哪位大侠能把完整的代码写出来呀最好是可以添加、删除和显示表格的）<%@pagecontentTypetexthtmlpageEn ... [详细]

蜡笔小新 2024-09-27 09:04:05
config
大数据学习环境搭建

大数据学习环境安装关于防火墙centos7使用的是firewalld，centos之前使用的是iptablesCentOS7关闭防火墙查看防火墙状态sudosy ... [详细]

蜡笔小新 2024-09-26 19:53:46
int
关于初学PHP时的知识积累总结【PHP】

后端开发|php教程PHP,知识积累后端开发-php教程PHP基础A、初识PHPPHP是与HTML混合使用的嵌入式语言。1、PHP标记默认标记短标记，需在php.ini中将shor ... [详细]

蜡笔小新 2024-09-25 19:33:38
callback
【从零到壹】Koa 从理解到实现

【从零到壹】Koa从理解到实现-【点击查看文中的相关源码】根据官网的介绍，Koa是一个新的Web框架，致力于成为Web应用和API开发领域中的一个更小、更富有表现力和更健壮的基石。 ... [详细]

蜡笔小新 2024-09-25 17:10:35
config
在云服务器中搭建Jupyter Notebook环境

目录前言二、JupyterNotebook搭建步骤1.云服务器准备2.安装Python及pip3.安装JupyterNotebook4.运行JupyterNoteboo ... [详细]

蜡笔小新 2024-09-25 13:24:52
int
深入理解Kafka服务端请求队列中请求的处理

本文深入分析了Kafka服务端请求队列中请求的处理过程，详细介绍了请求的封装和放入请求队列的过程，以及处理请求的线程池的创建和容量设置。通过场景分析、图示说明和源码分析，帮助读者更好地理解Kafka服务端的工作原理。 ... [详细]

蜡笔小新 2023-12-12 16:14:59

mobiledu2502884357

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章