def METHOD_NAME(
    self, config: Mapping[str, Any], configured_catalog: ConfiguredAirbyteCatalog, input_messages: Iterable[AirbyteMessage]
) -> Iterable[AirbyteMessage]:
    config_model = ConfigModel.parse_obj(config)
    self._init_indexer(config_model)
    self.processor = DocumentProcessor(config_model.processing, configured_catalog, max_metadata_size=self.indexer.max_metadata_size)
    batcher = Batcher(BATCH_SIZE, lambda batch: self._process_batch(batch))
    self.indexer.pre_sync(configured_catalog)
    for message in input_messages:
        if message.type == Type.STATE:
            # Emitting a state message indicates that all records which came before it have been written to the destination. So we flush
            # the queue to ensure writes happen, then output the state message to indicate it's safe to checkpoint state
            batcher.flush()
            yield message
        elif message.type == Type.RECORD:
            batcher.add(message.record)
    batcher.flush()
    yield from self.indexer.post_sync()