Address feedback

sea-streamer-kafka/src/producer.rs

@@ -4,9 +4,10 @@ use crate::{
     cluster::cluster_uri, impl_into_string, stream_err, BaseOptionKey, KafkaConnectOptions,
     KafkaErr, KafkaResult, DEFAULT_TIMEOUT,
 };
+pub use rdkafka::producer::FutureRecord;
 use rdkafka::{
     config::ClientConfig,
-    producer::{DeliveryFuture, FutureRecord as RawPayload, Producer as ProducerTrait},
+    producer::{DeliveryFuture, Producer as ProducerTrait},
 };
 pub use rdkafka::{consumer::ConsumerGroupMetadata, TopicPartitionList};
 use sea_streamer_runtime::spawn_blocking;
@@ -88,7 +89,7 @@ impl Producer for KafkaProducer {
     fn send_to<S: Buffer>(&self, stream: &StreamKey, payload: S) -> KafkaResult<Self::SendFuture> {
         let fut = self
             .get()
-            .send_result(RawPayload::<str, [u8]>::to(stream.name()).payload(payload.as_bytes()))
+            .send_result(FutureRecord::<str, [u8]>::to(stream.name()).payload(payload.as_bytes()))
             .map_err(|(err, _raw)| stream_err(err))?;
 
         Ok(SendFuture {
@@ -134,11 +135,8 @@ impl KafkaProducer {
             .expect("Producer is still inside a transaction, please await the future")
     }
 
-    /// Send a record to a stream
-    pub fn send_record<K, P>(
-        &self,
-        record: rdkafka::producer::FutureRecord<K, P>,
-    ) -> KafkaResult<SendFuture>
+    /// Send a `FutureRecord` to a stream
+    pub fn send_record<K, P>(&self, record: FutureRecord<K, P>) -> KafkaResult<SendFuture>
     where
         K: rdkafka::message::ToBytes + ?Sized,
         P: rdkafka::message::ToBytes + ?Sized,

sea-streamer-kafka/tests/consumer.rs

@@ -51,7 +51,6 @@ async fn main() -> anyhow::Result<()> {
             .partition(0)
             .payload(&payload);
         let receipt = producer.send_record(record)?.await?;
-        println!("Receipt: {:?}", receipt);
         assert_eq!(receipt.stream_key(), &topic);
         assert_eq!(receipt.sequence(), &i);
         assert_eq!(receipt.shard_id(), &zero);
@@ -115,6 +114,15 @@ async fn main() -> anyhow::Result<()> {
     assert_eq!(seq, [7, 8, 9]);
     println!("Seek stream ... ok");
 
+    let seq = consume(&mut consumer, 10).await;
+    // this should continue from 10
+    assert_eq!(seq, [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]);
+    println!("Resume ... ok");
+
+    // commit up to 19
+    consumer.commit(&topic, &zero, &19).await?;
+    println!("Commit ... ok");
+
     async fn consume(consumer: &mut KafkaConsumer, num: usize) -> Vec<usize> {
         consumer
             .stream()