schema

class BaseTaskOperateMsg(Record):
    task_ids = Array(Integer())


class TaskOffline2OnlineMsg(BaseTaskOperateMsg):
    pass

producer

client = pulsar.Client("pulsar://localhost:6650")
producer = client.create_producer(
    topic="taskOffline2Online1",
    schema=JsonSchema(TaskOffline2OnlineMsg),
)
msg = TaskOffline2OnlineMsg(task_ids=[1, 2])
producer.send(msg)
client.close()

consumer

client = pulsar.Client("pulsar://localhost:6650")


consumer = client.subscribe("taskOffline2Online1", "sub1", schema=JsonSchema(TaskOffline2OnlineMsg))

while True:
    msg = consumer.receive()
    try:
        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
        print(msg.value())
        # Acknowledge successful processing of the message
        consumer.acknowledge(msg)
    except Exception:
        # Message failed to be processed
        consumer.negative_acknowledge(msg)
    break

when using TaskOffline2OnlineMsg, i cant get task_ids.