Add missing mock APIs

Author: scanterog

rdkafka-sys/src/types.rs

@@ -5,6 +5,8 @@ use std::error::Error;
 use std::ffi::CStr;
 use std::fmt;
 
+use num_enum::IntoPrimitive;
+
 use crate::bindings;
 use crate::helpers;
 
@@ -477,6 +479,72 @@ impl fmt::Display for RDKafkaErrorCode {
 
 impl Error for RDKafkaErrorCode {}
 
+/// Native rdkafka ApiKeys / protocol requests
+#[derive(Debug, Clone, Copy, PartialEq, Eq, IntoPrimitive)]
+#[repr(i16)]
+#[non_exhaustive]
+pub enum RDKafkaApiKey {
+    Produce = 0,
+    Fetch = 1,
+    ListOffsets = 2,
+    Metadata = 3,
+    LeaderAndIsr = 4,
+    StopReplica = 5,
+    UpdateMetadata = 6,
+    ControlledShutdown = 7,
+    OffsetCommit = 8,
+    OffsetFetch = 9,
+    FindCoordinator = 10,
+    JoinGroup = 11,
+    Heartbeat = 12,
+    LeaveGroup = 13,
+    SyncGroup = 14,
+    DescribeGroups = 15,
+    ListGroups = 16,
+    SaslHandshake = 17,
+    ApiVersion = 18,
+    CreateTopics = 19,
+    DeleteTopics = 20,
+    DeleteRecords = 21,
+    InitProducerId = 22,
+    OffsetForLeaderEpoch = 23,
+    AddPartitionsToTxn = 24,
+    AddOffsetsToTxn = 25,
+    EndTxn = 26,
+    WriteTxnMarkers = 27,
+    TxnOffsetCommit = 28,
+    DescribeAcls = 29,
+    CreateAcls = 30,
+    DeleteAcls = 31,
+    DescribeConfigs = 32,
+    AlterConfigs = 33,
+    AlterReplicaLogDirs = 34,
+    DescribeLogDirs = 35,
+    SaslAuthenticate = 36,
+    CreatePartitions = 37,
+    CreateDelegationToken = 38,
+    RenewDelegationToken = 39,
+    ExpireDelegationToken = 40,
+    DescribeDelegationToken = 41,
+    DeleteGroups = 42,
+    ElectLeaders = 43,
+    IncrementalAlterConfigs = 44,
+    AlterPartitionReassignments = 45,
+    ListPartitionReassignments = 46,
+    OffsetDelete = 47,
+    DescribeClientQuotas = 48,
+    AlterClientQuotas = 49,
+    DescribeUserScramCredentials = 50,
+    AlterUserScramCredentials = 51,
+    Vote = 52,
+    BeginQuorumEpoch = 53,
+    EndQuorumEpoch = 54,
+    DescribeQuorum = 55,
+    AlterIsr = 56,
+    UpdateFeatures = 57,
+    Envelope = 58,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/mocking.rs

@@ -140,6 +140,40 @@ where
         bootstrap.to_string_lossy().to_string()
     }
 
+    /// Clear the cluster's error state for the given ApiKey.
+    pub fn clear_request_errors(&self, api_key: RDKafkaApiKey) {
+        unsafe { rdsys::rd_kafka_mock_clear_request_errors(self.mock_cluster, api_key.into()) }
+    }
+
+    /// Push errors onto the cluster's error stack for the given ApiKey.
+    ///
+    /// The protocol requests matching the given ApiKey will fail with the
+    /// provided error code and removed from the stack, starting with
+    /// the first error code, then the second, etc.
+    ///
+    /// Passing RD_KAFKA_RESP_ERR__TRANSPORT will make the mock broker
+    /// disconnect the client which can be useful to trigger a disconnect
+    /// on certain requests.
+    pub fn request_errors(&self, api_key: RDKafkaApiKey, errors: &[RDKafkaRespErr]) {
+        unsafe {
+            rdsys::rd_kafka_mock_push_request_errors_array(
+                self.mock_cluster,
+                api_key.into(),
+                errors.len(),
+                errors.as_ptr(),
+            )
+        }
+    }
+
+    /// Set the topic error to return in protocol requests.
+    ///
+    /// Currently only used for TopicMetadataRequest and AddPartitionsToTxnRequest.
+    pub fn topic_error(&self, topic: &str, error: RDKafkaRespErr) -> KafkaResult<()> {
+        let topic_c = CString::new(topic)?;
+        unsafe { rdsys::rd_kafka_mock_topic_set_error(self.mock_cluster, topic_c.as_ptr(), error) }
+        Ok(())
+    }
+
     /// Create a topic
     ///
     /// This is an alternative to automatic topic creation as performed by the client itself.
@@ -320,6 +354,31 @@ where
             }
         }
     }
+
+    /// Set the allowed ApiVersion range for the given ApiKey.
+    ///
+    /// Set min_version and max_version to `None` to disable the API completely.
+    /// max_version MUST not exceed the maximum implemented value.
+    pub fn apiversion(
+        &self,
+        api_key: RDKafkaApiKey,
+        min_version: Option<i16>,
+        max_version: Option<i16>,
+    ) -> KafkaResult<()> {
+        let min_version = min_version.unwrap_or(-1);
+        let max_version = max_version.unwrap_or(-1);
+
+        return_mock_op! {
+            unsafe {
+                rdsys::rd_kafka_mock_set_apiversion(
+                    self.mock_cluster,
+                    api_key.into(),
+                    min_version,
+                    max_version,
+                )
+            }
+        }
+    }
 }
 
 impl<'c, C> Drop for MockCluster<'c, C>