Repository: risingwavelabs/risingwave
File: src/connector/src/source/pulsar/source/reader.rs (Rust)

Code before:

        PulsarConsumeStream {
            source_ctx: self.source_ctx,
            split_id: self.split_id,
            pulsar_reader: self.consumer,
            ack_rx,
            topic: self.split.topic.to_string(),
        }
    }
}

struct PulsarConsumeStream {
    source_ctx: SourceContextRef,
    split_id: SplitId,
    pulsar_reader: Consumer<Vec<u8>, TokioExecutor>,
    ack_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
    topic: String,
}

impl PulsarConsumeStream {
    fn do_ack(&mut self, message_id_bytes: Vec<u8>) {
        let message_id = match MessageIdData::decode(message_id_bytes.as_slice()) {
            Ok(message_id) => message_id,
            Err(e) => {
                tracing::warn!(
                    error=?e, "meet error when decode message id, skip ack"
                );
                return;
            }
        };
        let topic = self.topic.clone();
        tracing::info!(
            "ack message id: {:?} from channel {}",
            message_id,
            build_pulsar_ack_channel_id(&self.source_ctx.source_id, &self.split_id)
        );
        #[cfg(not(madsim))]
        {
            // FIXME: madsim does not support block_on and block_in_place
            let _ = tokio::task::block_in_place(move || {
                tokio::runtime::Handle::current().block_on(async {
                    self.pulsar_reader
                        .cumulative_ack_with_id(&topic, message_id)
                        .await
                })
            })
            .map_err(|e| {
                tracing::warn!(
                    error=?e, "meet error when ack message"
                )
            });

Review comment (suggestion):
```suggestion
                    error=%e.as_report(), "meet error when decode message id, skip ack"
```

Code after:
    }
}

impl PulsarBrokerReader {
    #[try_stream(ok = Vec<SourceMessage>, error = crate::error::ConnectorError)]
    async fn into_data_stream(mut self) {
        let max_chunk_size = self.source_ctx.source_ctrl_opts.chunk_size;
        let mut already_read_offset = self.already_read_offset.take();
        #[for_await]
        for msgs in self.into_stream().await.ready_chunks(max_chunk_size) {
            let msgs = msgs
                .into_iter()
                .collect::<Result<Vec<Message<Vec<u8>>>, _>>()?;
            let mut res = Vec::with_capacity(msgs.len());
            for msg in msgs {
                if let Some(PulsarFilterOffset {
                    ledger_id,
                    entry_id,
                    batch_index,
                }) = already_read_offset
                {
                    let message_id = msg.message_id();

                    // If we have a previously stored offset, check that the messages we're consuming
                    // appear after. Note that entry IDs and batch indexes are only monotonically increasing
                    // within a BookKeeper ledger, so we must check that the ledger IDs match. If they don't match,
                    // messages are coming from a new ledger and we can safely ignore the stored offset.
                    if message_id.ledger_id == ledger_id
                        && message_id.entry_id <= entry_id
                        && message_id.batch_index <= batch_index
                    {
                        tracing::info!(
                            "skipping message with entry_id: {}, batch_index: {:?} as expected offset after entry_id {} batch_index {:?}",
                            message_id.entry_id,
                            message_id.batch_index,
                            entry_id,
                            batch_index
                        );
                        continue;
                    } else {
                        already_read_offset = None;
                    }
                }

                let msg = SourceMessage::from(msg);
                res.push(msg);
            }
            yield res;
        }
    }