def METHOD_NAME(
    self,
    config: Mapping[str, Any],
    configured_catalog: ConfiguredAirbyteCatalog,
    input_messages: Iterable[AirbyteMessage],
) -> Iterable[AirbyteMessage]:
    """Reads the input stream of messages, config, and catalog to write data to the destination.
    This method returns an iterable (typically a generator of AirbyteMessages via yield) containing state messages received in the
    input message stream. Outputting a state message means that every AirbyteRecordMessage which came before it has been successfully
    persisted to the destination. This is used to ensure fault tolerance in the case that a sync fails before fully completing,
    then the source is given the last state message output from this method as the starting point of the next sync.
    :param config: dict of JSON configuration matching the configuration declared in spec.json. Current format:
        {
            'api_host': '<api_host_url, e.g. https://api.cumul.io>',
            'api_key': '<api_key>',
            'api_token': '<api_token>'
        }
    :param configured_catalog: schema of the data being received and how it should be persisted in the destination.
    :param input_messages: stream of input messages received from the source.
    :return: Iterable of AirbyteStateMessages wrapped in AirbyteMessage structs.
    """
    writer = CumulioWriter(config, configured_catalog, logger)
    for configured_stream in configured_catalog.streams:
        # Cumul.io does not support removing all data from an existing dataset, and removing the dataset itself will break existing
        # dashboards built on top of it.
        # Instead, the connector will make sure to push the first batch of data as a "replace" action: this will cause all existing data
        # to be replaced with the first batch of data. All next batches will be pushed as an "append" action.
        if configured_stream.destination_sync_mode == DestinationSyncMode.overwrite:
            writer.delete_stream_entries(configured_stream.stream.name)
    for message in input_messages:
        if message.type == Type.STATE:
            # Yielding a state message indicates that all records which came before it have been written to the destination.
            # We flush all write buffers in the writer, and then output the state message itself.
            writer.flush_all()
            yield message
        elif message.type == Type.RECORD:
            record = message.record
            assert record is not None
            assert record.stream is not None
            assert record.data is not None
            writer.queue_write_operation(record.stream, record.data)
        else:
            # ignore other message types for now
            continue
    # Make sure to flush any records still in the queue
    writer.flush_all()