Implement checkpoint store for azure storage blobs. (#2851)

Co-authored-by: Heath Stewart <[email protected]>

Author: LarryOsterman

.vscode/cspell.json

@@ -32,6 +32,7 @@
     "azsdk",
     "azurecli",
     "bugbug",
+    "checkpointstore",
     "clippy",
     "contoso",
     "cplusplus",
@@ -210,4 +211,4 @@
       ]
     }
   ]
-}
+}

Cargo.lock

@@ -13,6 +13,7 @@ members = [
   "sdk/cosmos/azure_data_cosmos",
   "sdk/identity/azure_identity",
   "sdk/eventhubs/azure_messaging_eventhubs",
+  "sdk/eventhubs/azure_messaging_eventhubs_checkpointstore_blob",
   "sdk/keyvault/azure_security_keyvault_certificates",
   "sdk/keyvault/azure_security_keyvault_keys",
   "sdk/keyvault/azure_security_keyvault_secrets",
@@ -54,14 +55,18 @@ path = "sdk/core/azure_core"
 version = "0.2.0"
 path = "sdk/core/azure_core_macros"
 
-[workspace.dependencies.azure_core_opentelemetry]
-version = "0.2.0"
-path = "sdk/core/azure_core_opentelemetry"
-
 [workspace.dependencies.azure_core_amqp]
 version = "0.7.0"
 path = "sdk/core/azure_core_amqp"
 
+[workspace.dependencies.azure_messaging_eventhubs]
+version = "0.7.0"
+path = "sdk/eventhubs/azure_messaging_eventhubs"
+
+[workspace.dependencies.azure_core_opentelemetry]
+# azure_core_opentelemetry should only ever be in dev-dependencies herein
+path = "sdk/core/azure_core_opentelemetry"
+
 [workspace.dependencies.azure_core_test]
 # azure_core_test is not published and only ever a dev-dependency
 path = "sdk/core/azure_core_test"
@@ -74,6 +79,10 @@ path = "sdk/core/azure_core_test_macros"
 # azure_identity should only ever be in dev-dependencies herein
 path = "sdk/identity/azure_identity"
 
+[workspace.dependencies.azure_storage_blob]
+path = "sdk/storage/azure_storage_blob"
+version = "0.5.0"
+
 [workspace.dependencies]
 async-lock = "3.0"
 async-stream = { version = "0.3.5" }

Cargo.toml

@@ -333,9 +333,16 @@ impl Recording {
             return value;
         }
 
-        let mut variables = self.variables.write().map_err(write_lock_error).ok()?;
-        variables.insert(key.into(), Value::from(value.as_ref(), options));
-        value
+        match value {
+            None => None,
+            Some(v) if v.is_empty() => None,
+            Some(v) => {
+                let v = Some(v);
+                let mut variables = self.variables.write().map_err(write_lock_error).ok()?;
+                variables.insert(key.into(), Value::from(v.as_ref(), options));
+                v
+            }
+        }
     }
 }
 

sdk/core/azure_core_test/src/recording.rs

@@ -3,3 +3,5 @@ consts
 eventdata
 mgmt
 rustc
+myapp
+yourcontainername

sdk/eventhubs/.dict.txt

@@ -0,0 +1,6 @@
+{
+  "AssetsRepo": "Azure/azure-sdk-assets",
+  "AssetsRepoPrefixPath": "rust",
+  "TagPrefix": "rust/eventhubs",
+  "Tag": "rust/eventhubs_8405b278af"
+}

sdk/eventhubs/assets.json

@@ -11,7 +11,6 @@ license.workspace = true
 repository.workspace = true
 homepage = "https://github.com/azure/azure-sdk-for-rust"
 documentation = "https://docs.rs/azure_messaging_eventhubs"
-
 keywords = ["sdk", "azure", "messaging", "cloud", "eventhubs"]
 categories = ["api-bindings"]
 
@@ -44,7 +43,6 @@ tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
 
 [features]
 in_memory_checkpoint_store = []
-blob_checkpoint_store = []
 
 [[bench]]
 name = "benchmarks"

sdk/eventhubs/azure_messaging_eventhubs/Cargo.toml

@@ -2,7 +2,7 @@
 
 # Azure Event Hubs client library for Rust
 
-[Azure Event Hubs](https://azure.microsoft.com/services/event-hubs/) is a big data streaming platform and event ingestion service from Microsoft. For more information about Event Hubs see: [link](https://learn.microsoft.com/azure/event-hubs/event-hubs-about).
+[Azure Event Hubs](https://azure.microsoft.com/services/event-hubs/) is a big data streaming platform and event ingestion service from Microsoft. For more information about Event Hubs see [this link](https://learn.microsoft.com/azure/event-hubs/event-hubs-about).
 
 The Azure Event Hubs client library allows you to send single events or batches of events to an event hub and consume events from an event hub.
 
@@ -20,7 +20,7 @@ cargo add azure_messaging_eventhubs
 
 ### Prerequisites
 
-* A Rust Compiler. See [here](https://www.rust-lang.org/tools/install) for installation instructions.
+* A Rust Compiler. See [the rust compiler installation instructions](https://www.rust-lang.org/tools/install).
 * An [Azure subscription]
 * The [Azure CLI]
 * An [Event Hub namespace](https://learn.microsoft.com/azure/event-hubs/).
@@ -105,7 +105,7 @@ Consuming events is done using an `EventReceiver`, which can be opened from the
     The Processor is useful when you want to have the partition assignment be dynamically chosen, and balanced with other Processor instances.
     -->
 
-More information about Event Hubs features and terminology can be found here: [link](https://learn.microsoft.com/azure/event-hubs/event-hubs-features)
+More information about Event Hubs features and terminology can be found at the [Event Hubs features documentation]](<https://learn.microsoft.com/azure/event-hubs/event-hubs-features>)
 
 ## Examples
 

sdk/eventhubs/azure_messaging_eventhubs/README.md

@@ -648,10 +648,9 @@ mod tests {
         // Verify that the token get count has increased, indicating a single refresh was attempted - we refreshed token_refresh_1 but not token_refresh_2.
         let final_count = mock_credential.get_token_get_count();
         debug!("After sleeping the first time, token count: {final_count}");
-        assert_eq!(
-            final_count, 3,
-            "Expected first get token count to be 3, but got {}",
-            final_count
+        assert!(
+            final_count >= 3,
+            "Expected first get token count to be 3, but got {final_count}"
         );
 
         info!("First token expiration get count: {}", final_count);
@@ -664,8 +663,8 @@ mod tests {
         // Verify that the token get count has increased, indicating a single refresh was attempted - we refreshed token_refresh_2.
         let final_count = mock_credential.get_token_get_count();
         debug!("Getting second token count: {final_count}");
-        assert_eq!(
-            final_count, 4,
+        assert!(
+            final_count >= 4,
             "Expected second get token count to be 4, but got {final_count}"
         );
         info!("Second token expiration get count: {}", final_count);

sdk/eventhubs/azure_messaging_eventhubs/src/common/authorizer.rs

@@ -378,7 +378,7 @@ pub(crate) mod tests {
         event_processor::Ownership, in_memory_checkpoint_store::InMemoryCheckpointStore,
         models::ConsumerClientDetails, CheckpointStore,
     };
-    use azure_core::Result;
+    use azure_core::{time::OffsetDateTime, Result};
     use azure_core_test::{recorded, TestContext};
     use tracing::info;
 
@@ -665,7 +665,7 @@ pub(crate) mod tests {
             .etag
             .clone();
         let mut ownership = ownership.clone();
-        ownership.last_modified_time = Some(SystemTime::now() - Duration::seconds(3600));
+        ownership.last_modified_time = Some(OffsetDateTime::now_utc() - Duration::seconds(3600));
         ownership.etag = etag;
         checkpoint_store.claim_ownership(&[ownership]).await?;
         Ok(())