This message was deleted.

The error message you're seeing is due to an assertion in your code that checks if the number of state messages is greater than or equal to the number of streams in your configured catalog. In your case, it seems like you have 6 streams but only 1 state message, which is causing the assertion to fail. The

AirbyteStateMessage

is used to persist state between sync attempts, especially when a prior attempt fails. This allows subsequent sync attempts of a job to pick up from the last checkpoint of the previous one. For resumable full refresh syncs, state is passed in between sync attempts, but deleted at the beginning of new sync jobs ([source](https://docs.airbyte.com/connector-development/cdk-python/resumable-full-refresh-stream#airbytestatemessage)). To resolve this error, you need to ensure that at least one state message is emitted for each stream in your configured catalog. This could be done in the

read

function of your source connector where you emit

AirbyteRecordMessage

for each record in your stream. After reading the data and before sending it to the destination connector, you can emit an

AirbyteStateMessage

for each stream ([source](https://discuss.airbyte.io/t/1456)). Here is an example of how you might emit an

AirbyteRecordMessage

and an

AirbyteStateMessage

for each stream:

python for stream in streams:     for record in stream:         record_message = AirbyteRecordMessage(stream=stream, data=record, emitted_at=int(datetime.now().timestamp()) * 1000)         yield AirbyteMessage(type=Type.RECORD, record=record_message)         state_message = AirbyteStateMessage(data={...})  # replace with your state data         yield AirbyteMessage(type=Type.STATE, state=state_message)

Please replace

{...}

with the actual state data for each stream. The state data is typically a dictionary that contains information about the state of the stream, such as the last processed record or the current position in the stream. Remember, the state message enables the Source to emit checkpoints while replicating data. These checkpoints mean that if replication fails before completion, the next sync is able to start from the last checkpoint instead of returning to the beginning of the previous sync ([source](https://docs.airbyte.com/understanding-airbyte/airbyte-protocol#airbytestatemessage)). If you're still having trouble, please provide more information about your source connector and how you're handling state messages.