Merge pull request #191 from influxdata/crepererum/issue190

feat: implement topic deletion

Author: kodiakhq[bot]

src/client/controller.rs

@@ -14,8 +14,8 @@ use crate::{
     messenger::RequestError,
     protocol::{
         error::Error as ProtocolError,
-        messages::{CreateTopicRequest, CreateTopicsRequest},
-        primitives::{Int16, Int32, String_},
+        messages::{CreateTopicRequest, CreateTopicsRequest, DeleteTopicsRequest},
+        primitives::{Array, Int16, Int32, String_},
     },
     throttle::maybe_throttle,
     validation::ExactlyOne,
@@ -103,6 +103,57 @@ impl ControllerClient {
         Ok(())
     }
 
+    /// Delete a topic
+    pub async fn delete_topic(
+        &self,
+        name: impl Into<String> + Send,
+        timeout_ms: i32,
+    ) -> Result<()> {
+        let request = &DeleteTopicsRequest {
+            topic_names: Array(Some(vec![String_(name.into())])),
+            timeout_ms: Int32(timeout_ms),
+            tagged_fields: None,
+        };
+
+        maybe_retry(&self.backoff_config, self, "delete_topic", || async move {
+            let (broker, gen) = self
+                .get()
+                .await
+                .map_err(|e| ErrorOrThrottle::Error((e, None)))?;
+            let response = broker
+                .request(request)
+                .await
+                .map_err(|e| ErrorOrThrottle::Error((e.into(), Some(gen))))?;
+
+            maybe_throttle(response.throttle_time_ms)?;
+
+            let topic = response
+                .responses
+                .exactly_one()
+                .map_err(|e| ErrorOrThrottle::Error((Error::exactly_one_topic(e), Some(gen))))?;
+
+            match topic.error {
+                None => Ok(()),
+                Some(protocol_error) => Err(ErrorOrThrottle::Error((
+                    Error::ServerError {
+                        protocol_error,
+                        error_message: topic.error_message.and_then(|s| s.0),
+                        request: RequestContext::Topic(topic.name.0),
+                        response: None,
+                        is_virtual: false,
+                    },
+                    Some(gen),
+                ))),
+            }
+        })
+        .await?;
+
+        // Refresh the cache now there is definitely a new topic to observe.
+        let _ = self.brokers.refresh_metadata().await;
+
+        Ok(())
+    }
+
     /// Retrieve the broker ID of the controller
     async fn get_controller_id(&self) -> Result<i32> {
         // Request an uncached, fresh copy of the metadata.

src/protocol/messages/delete_topics.rs

@@ -0,0 +1,173 @@
+use std::io::{Read, Write};
+
+use crate::protocol::{
+    api_key::ApiKey,
+    api_version::{ApiVersion, ApiVersionRange},
+    error::Error,
+    messages::{
+        read_compact_versioned_array, read_versioned_array, ReadVersionedError, ReadVersionedType,
+        RequestBody, WriteVersionedError, WriteVersionedType,
+    },
+    primitives::{
+        Array, CompactArrayRef, CompactNullableString, CompactString, CompactStringRef, Int16,
+        Int32, String_, TaggedFields,
+    },
+    traits::{ReadType, WriteType},
+};
+
+#[derive(Debug)]
+pub struct DeleteTopicsRequest {
+    /// The names of the topics to delete.
+    pub topic_names: Array<String_>,
+
+    /// The length of time in milliseconds to wait for the deletions to complete.
+    pub timeout_ms: Int32,
+
+    /// The tagged fields.
+    ///
+    /// Added in version 4.
+    pub tagged_fields: Option<TaggedFields>,
+}
+
+impl RequestBody for DeleteTopicsRequest {
+    type ResponseBody = DeleteTopicsResponse;
+
+    const API_KEY: ApiKey = ApiKey::DeleteTopics;
+
+    /// Enough for now.
+    const API_VERSION_RANGE: ApiVersionRange =
+        ApiVersionRange::new(ApiVersion(Int16(0)), ApiVersion(Int16(5)));
+
+    const FIRST_TAGGED_FIELD_IN_REQUEST_VERSION: ApiVersion = ApiVersion(Int16(4));
+}
+
+impl<W> WriteVersionedType<W> for DeleteTopicsRequest
+where
+    W: Write,
+{
+    fn write_versioned(
+        &self,
+        writer: &mut W,
+        version: ApiVersion,
+    ) -> Result<(), WriteVersionedError> {
+        let v = version.0 .0;
+        assert!(v <= 5);
+
+        if v >= 4 {
+            if let Some(topic_names) = self.topic_names.0.as_ref() {
+                let topic_names: Vec<_> = topic_names
+                    .iter()
+                    .map(|name| CompactStringRef(name.0.as_str()))
+                    .collect();
+                CompactArrayRef(Some(&topic_names)).write(writer)?;
+            } else {
+                CompactArrayRef::<CompactStringRef<'_>>(None).write(writer)?;
+            }
+        } else {
+            self.topic_names.write(writer)?;
+        };
+
+        self.timeout_ms.write(writer)?;
+
+        if v >= 4 {
+            match self.tagged_fields.as_ref() {
+                Some(tagged_fields) => {
+                    tagged_fields.write(writer)?;
+                }
+                None => {
+                    TaggedFields::default().write(writer)?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+pub struct DeleteTopicsResponse {
+    /// The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the
+    /// request did not violate any quota.
+    ///
+    /// Added in version 1.
+    pub throttle_time_ms: Option<Int32>,
+
+    /// The results for each topic we tried to delete.
+    pub responses: Vec<DeleteTopicsResponseTopic>,
+
+    /// The tagged fields.
+    ///
+    /// Added in version 4.
+    pub tagged_fields: Option<TaggedFields>,
+}
+
+impl<R> ReadVersionedType<R> for DeleteTopicsResponse
+where
+    R: Read,
+{
+    fn read_versioned(reader: &mut R, version: ApiVersion) -> Result<Self, ReadVersionedError> {
+        let v = version.0 .0;
+        assert!(v <= 5);
+
+        let throttle_time_ms = (v >= 1).then(|| Int32::read(reader)).transpose()?;
+        let responses = if v >= 4 {
+            read_compact_versioned_array(reader, version)?.unwrap_or_default()
+        } else {
+            read_versioned_array(reader, version)?.unwrap_or_default()
+        };
+        let tagged_fields = (v >= 4).then(|| TaggedFields::read(reader)).transpose()?;
+
+        Ok(Self {
+            throttle_time_ms,
+            responses,
+            tagged_fields,
+        })
+    }
+}
+
+#[derive(Debug)]
+pub struct DeleteTopicsResponseTopic {
+    /// The topic name.
+    pub name: String_,
+
+    /// The error code, or 0 if there was no error.
+    pub error: Option<Error>,
+
+    /// The error message, or null if there was no error.
+    ///
+    /// Added in version 5.
+    pub error_message: Option<CompactNullableString>,
+
+    /// The tagged fields.
+    ///
+    /// Added in version 4.
+    pub tagged_fields: Option<TaggedFields>,
+}
+
+impl<R> ReadVersionedType<R> for DeleteTopicsResponseTopic
+where
+    R: Read,
+{
+    fn read_versioned(reader: &mut R, version: ApiVersion) -> Result<Self, ReadVersionedError> {
+        let v = version.0 .0;
+        assert!(v <= 5);
+
+        let name = if v >= 4 {
+            String_(CompactString::read(reader)?.0)
+        } else {
+            String_::read(reader)?
+        };
+        let error = Error::new(Int16::read(reader)?.0);
+        let error_message = (v >= 5)
+            .then(|| CompactNullableString::read(reader))
+            .transpose()?;
+        let tagged_fields = (v >= 4).then(|| TaggedFields::read(reader)).transpose()?;
+
+        Ok(Self {
+            name,
+            error,
+            error_message,
+            tagged_fields,
+        })
+    }
+}

src/protocol/messages/mod.rs

@@ -24,6 +24,8 @@ mod create_topics;
 pub use create_topics::*;
 mod delete_records;
 pub use delete_records::*;
+mod delete_topics;
+pub use delete_topics::*;
 mod fetch;
 pub use fetch::*;
 mod header;

tests/client.rs

@@ -46,18 +46,26 @@ async fn test_topic_crud() {
             }
         }
     }
-    let new_topic = format!("{}{}", prefix, max_id + 1);
-    controller_client
-        .create_topic(&new_topic, 2, 1, 5_000)
-        .await
-        .unwrap();
+
+    // create two topics
+    let mut new_topics = vec![];
+    for offset in 1..=2 {
+        let new_topic = format!("{}{}", prefix, max_id + offset);
+        controller_client
+            .create_topic(&new_topic, 2, 1, 5_000)
+            .await
+            .unwrap();
+        new_topics.push(new_topic);
+    }
 
     // might take a while to converge
     tokio::time::timeout(TEST_TIMEOUT, async {
         loop {
             let topics = client.list_topics().await.unwrap();
-            let topic = topics.iter().find(|t| t.name == new_topic);
-            if topic.is_some() {
+            if new_topics
+                .iter()
+                .all(|new_topic| topics.iter().any(|t| &t.name == new_topic))
+            {
                 return;
             }
 
@@ -67,8 +75,9 @@ async fn test_topic_crud() {
     .await
     .unwrap();
 
+    // topic already exists
     let err = controller_client
-        .create_topic(&new_topic, 1, 1, 5_000)
+        .create_topic(&new_topics[0], 1, 1, 5_000)
         .await
         .unwrap_err();
     match err {
@@ -78,6 +87,28 @@ async fn test_topic_crud() {
         } => {}
         _ => panic!("Unexpected error: {}", err),
     }
+
+    // delete one topic
+    controller_client
+        .delete_topic(&new_topics[0], 5_000)
+        .await
+        .unwrap();
+
+    // might take a while to converge
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        loop {
+            let topics = client.list_topics().await.unwrap();
+            if topics.iter().all(|t| t.name != new_topics[0])
+                && topics.iter().any(|t| t.name == new_topics[1])
+            {
+                return;
+            }
+
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
 }
 
 #[tokio::test]