Adding framework for perf testing, samples, README.md for JedisRedisCheckpointStore

* Adding framework for perf testing, samples, README.md for JedisRedisCheckpointStore

Author: anushkasingh16

eng/code-quality-reports/src/main/resources/revapi/revapi.json

@@ -365,6 +365,18 @@
           "new": "parameter <T> reactor.core.publisher.Mono<T> com.azure.spring.data.cosmos.core.ReactiveCosmosTemplate::insert(java.lang.String, ===T===, com.azure.cosmos.models.PartitionKey)",
           "parameterIndex": "1",
           "justification": "To support mono method chaining, without explicit typcast for upper bounded generics"
+        },
+        {
+          "regex": true,
+          "code": "java.class.externalClassExposedInAPI",
+          "new": "(interface|class|enum) redis\\.clients\\.jedis\\..*",
+          "justification": "To support the EventHubs JedisRedisCheckpointStore constructor"
+        },
+        {
+          "regex": true,
+          "code": "java.class.externalClassExposedInAPI",
+          "new": "class org\\.json\\.JSON(Array|Exception|Object|Pointer|PointerException|Tokener)",
+          "justification": "To support the EventHubs JedisRedisCheckpointStore constructor"
         }
       ]
     }

sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/README.md

@@ -1,13 +1,176 @@
-# Azure Event Hubs Checkpoint Store using Redis client library for Java
+# Azure Event Hubs Checkpoint Store client library for Java using the Jedis Client Library for Redis
+
+Azure Event Hubs Checkpoint Store can be used for storing checkpoints while processing events from Azure Event Hubs.
+This package makes use of Redis as a persistent store for maintaining checkpoints and partition ownership information.
+The `JedisRedisCheckpointStore` provided in this package can be plugged in to `EventProcessor`.
+
+[Source code][source_code]| [API reference documentation][api_documentation] | [Product
+documentation][event_hubs_product_docs] | [Samples][sample_examples]
 
 ## Getting started
 
+### Prerequisites
+
+- A [Java Development Kit (JDK)][jdk_link], version 8 or later.
+- [Maven][maven]
+- Microsoft Azure subscription
+    - You can create a free account at: [https://azure.microsoft.com](https://azure.microsoft.com)
+- Azure Event Hubs instance
+    - Step-by-step guide for [creating an Event Hub using the Azure Portal][event_hubs_create]
+- Azure Redis Cache or a suitable alternative Redis server
+    - Step-by-step guide for [creating a Redis Cache using the Azure Portal][redis_cache]
+
+### Include the package
+#### Include the BOM file
+
+Please include the azure-sdk-bom to your project to take dependency on the General Availability (GA) version of the library. In the following snippet, replace the {bom_version_to_target} placeholder with the version number.
+To learn more about the BOM, see the [AZURE SDK BOM README](https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/boms/azure-sdk-bom/README.md).
+
+```xml
+<dependencyManagement>
+    <dependencies>
+        <dependency>
+            <groupId>com.azure</groupId>
+            <artifactId>azure-sdk-bom</artifactId>
+            <version>{bom_version_to_target}</version>
+            <type>pom</type>
+            <scope>import</scope>
+        </dependency>
+    </dependencies>
+</dependencyManagement>
+```
+and then include the direct dependency in the dependencies section without the version tag as shown below.
+
+```xml
+<dependencies>
+  <dependency>
+    <groupId>com.azure</groupId>
+    <artifactId>azure-messaging-eventhubs-checkpointstore-jedis</artifactId>
+  </dependency>
+</dependencies>
+```
+
+#### Include direct dependency
+If you want to take dependency on a particular version of the library that is not present in the BOM,
+add the direct dependency to your project as follows.
+
+[//]: # ({x-version-update-start;com.azure:azure-messaging-eventhubs-checkpointstore-jedis;current})
+```xml
+<dependency>
+    <groupId>com.azure</groupId>
+    <artifactId>azure-messaging-eventhubs-checkpointstore-jedis</artifactId>
+    <version>1.0.0-beta.1</version>
+</dependency>
+```
+[//]: # ({x-version-update-end})
+
+### Authenticate the storage container client
+
+In order to create an instance of `JedisCheckpointStore`, a `JedisPool` object must be created. To make this `JedisPool` object, a hostname String and a primary key String are required. These can be used as shown below to create a `JedisPool` object.
+
+
 ## Key concepts
 
+Key concepts are explained in detail [here][key_concepts].
+
 ## Examples
+- [Create and run an instance of JedisRedisCheckpointStore][sample_jedis_client]
+- [Consume events from all Event Hub partitions][sample_event_processor]
+
+### Create an instance of JedisPool with Azure Redis Cache
+
+```java
+String hostname = "yourHostName.redis.cache.windows.net";
+
+String password = "<PRIMARY KEY FOR AZURE REDIS CACHE>";
+
+String name = "<NAME OF THE USER CLIENT>"; //this can also be a default value as the connection of Redis Cache is not dependent on this value
+
+JedisPool jedisPool = new JedisPool(poolConfig, hostname, port, 1000, 1000, password, Protocol.DEFAULT_DATABASE, name, true, null, null, null);
+```
+
+### Consume events using an Event Processor Client
+
+To consume events for all partitions of an Event Hub, you'll create an
+[`EventProcessorClient`][source_eventprocessorclient] for a specific consumer group. When an Event Hub is created, it
+provides a default consumer group that can be used to get started.
+
+The [`EventProcessorClient`][source_eventprocessorclient] will delegate processing of events to a callback function that you
+provide, allowing you to focus on the logic needed to provide value while the processor holds responsibility for
+managing the underlying consumer operations.
+
+In our example, we will focus on building the [`EventProcessor`][source_eventprocessorclient], use the
+[`JedisRedisCheckpointStore`][source_jedisredischeckpointstore], and a simple callback function to process the events
+received from the Event Hubs, writes to console and updates the checkpoint in Blob storage after each event.
+
+```java
+JedisPool jedisPool = new JedisPool(poolConfig, hostname, port, 1000, 1000, password, Protocol.DEFAULT_DATABASE, name, true, null, null, null);
+
+EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder()
+    .consumerGroup("<< CONSUMER GROUP NAME >>")
+    .connectionString("<< EVENT HUB CONNECTION STRING >>")
+    .checkpointStore(new JedisRedisCheckpointStore(jedisPool))
+    .processEvent(eventContext -> {
+        System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId() + " and "
+            + "sequence number of event = " + eventContext.getEventData().getSequenceNumber());
+    })
+    .processError(errorContext -> {
+        System.out.println("Error occurred while processing events " + errorContext.getThrowable().getMessage());
+    })
+    .buildEventProcessorClient();
+
+// This will start the processor. It will start processing events from all partitions.
+eventProcessorClient.start();
+
+// (for demo purposes only - adding sleep to wait for receiving events)
+TimeUnit.SECONDS.sleep(5);
+
+// When the user wishes to stop processing events, they can call `stop()`.
+eventProcessorClient.stop();
+```
 
 ## Troubleshooting
 
+### Enable client logging
+
+Azure SDK for Java offers a consistent logging story to help aid in troubleshooting application errors and expedite
+their resolution. The logs produced will capture the flow of an application before reaching the terminal state to help
+locate the root issue. View the [logging][logging] wiki for guidance about enabling logging.
+
+### Default SSL library
+
+All client libraries, by default, use the Tomcat-native Boring SSL library to enable native-level performance for SSL
+operations. The Boring SSL library is an uber jar containing native libraries for Linux / macOS / Windows, and provides
+better performance compared to the default SSL implementation within the JDK. For more information, including how to
+reduce the dependency size, refer to the [performance tuning][performance_tuning] section of the wiki.
+
 ## Next steps
 
+Get started by exploring the samples [here][samples_readme].
+
 ## Contributing
+
+If you would like to become an active contributor to this project please refer to our [Contribution
+Guidelines](https://github.com/Azure/azure-sdk-for-java/blob/main/CONTRIBUTING.md) for more information.
+
+<!-- Links -->
+[api_documentation]: https://azure.github.io/azure-sdk-for-java
+[event_hubs_create]: https://docs.microsoft.com/azure/event-hubs/event-hubs-create
+[event_hubs_product_docs]: https://docs.microsoft.com/azure/event-hubs/
+[java_8_sdk_javadocs]: https://docs.oracle.com/javase/8/docs/api/java/util/logging/package-summary.html
+[jdk_link]: https://docs.microsoft.com/java/azure/jdk/?view=azure-java-stable
+[key_concepts]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/README.md#key-concepts
+[logging]: https://github.com/Azure/azure-sdk-for-java/wiki/Logging-with-Azure-SDK
+[maven]: https://maven.apache.org/
+[performance_tuning]: https://github.com/Azure/azure-sdk-for-java/wiki/Performance-Tuning
+[redis_cache]: https://docs.microsoft.com/azure/azure-cache-for-redis/cache-configure
+[samples_readme]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis
+[//]: # ([sample_jedis_client]: ttps://github.com/Azure/azure-sdk-for-java/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/src/samples/java/com/azure/messaging/eventhubs/checkpointstore/jedis/JedisRedisCheckpointStoreSample.java)
+[//]: # ([sample_event_processor]: https://github.com/Azure/azure-sdk-for-java/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/src/samples/java/com/azure/messaging/eventhubs/checkpointstore/jedis/EventProcessorJedisRedisCheckpointStoreSample.java)
+[//]: # ([sample_examples]: https://github.com/Azure/azure-sdk-for-java/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/src/samples/java/com/azure/messaging/eventhubs/checkpointstore/jedis)
+[sample_jedis_client]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis
+[sample_event_processor]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis
+[sample_examples]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis
+[source_code]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis
+[source_eventprocessorclient]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java
+[source_jedisredischeckpointstore]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis

sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/pom.xml

@@ -78,6 +78,7 @@
       <groupId>org.mockito</groupId>
       <artifactId>mockito-inline</artifactId>
       <version>4.5.1</version> <!-- {x-version-update;org.mockito:mockito-inline;external_dependency} -->
+      <scope>test</scope>
     </dependency>
   </dependencies>
 

sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/src/main/java/com/azure/messaging/eventhubs/checkpointstore/jedis/JedisRedisCheckpointStore.java

@@ -34,7 +34,18 @@ public class JedisRedisCheckpointStore implements CheckpointStore {
     static final byte[] PARTITION_OWNERSHIP = "partitionOwnership".getBytes(StandardCharsets.UTF_8);
     private final JedisPool jedisPool;
 
-    JedisRedisCheckpointStore(JedisPool jedisPool) {
+    /**
+     * Constructor for JedisRedisCheckpointStore
+     *
+     * @param jedisPool a JedisPool object that creates a pool connected to the Azure Redis Cache
+     * @throws IllegalArgumentException thrown when JedisPool object supplied is null
+     */
+    public JedisRedisCheckpointStore(JedisPool jedisPool) throws IllegalArgumentException {
+        if (jedisPool == null) {
+            throw LOGGER.logExceptionAsError(Exceptions
+                .propagate(new IllegalArgumentException(
+                    "JedisPool object supplied to constructor is null.")));
+        }
         this.jedisPool = jedisPool;
     }
 
@@ -48,33 +59,40 @@ public class JedisRedisCheckpointStore implements CheckpointStore {
     public Flux<PartitionOwnership> claimOwnership(List<PartitionOwnership> requestedPartitionOwnerships) {
 
         return Flux.fromIterable(requestedPartitionOwnerships).handle(((partitionOwnership, sink) -> {
-
             String partitionId = partitionOwnership.getPartitionId();
             byte[] key = keyBuilder(partitionOwnership.getFullyQualifiedNamespace(), partitionOwnership.getEventHubName(), partitionOwnership.getConsumerGroup(), partitionId);
 
             try (Jedis jedis = jedisPool.getResource()) {
+
                 List<byte[]> keyInformation = jedis.hmget(key, PARTITION_OWNERSHIP);
                 byte[] currentPartitionOwnership = keyInformation.get(0);
 
                 if (currentPartitionOwnership == null) {
                     // if PARTITION_OWNERSHIP field does not exist for member we will get a null, and we must add the field
+                    Long lastModifiedTimeSeconds = Long.parseLong(jedis.time().get(0));
+                    partitionOwnership.setLastModifiedTime(lastModifiedTimeSeconds);
                     jedis.hset(key, PARTITION_OWNERSHIP, DEFAULT_SERIALIZER.serializeToBytes(partitionOwnership));
+                    sink.next(partitionOwnership);
+                    sink.complete();
                 } else {
                     // otherwise we have to change the ownership and "watch" the transaction
                     jedis.watch(key);
-
+                    Long lastModifiedTimeSeconds = Long.parseLong(jedis.time().get(0)) - jedis.objectIdletime(key);
+                    partitionOwnership.setLastModifiedTime(lastModifiedTimeSeconds);
+                    partitionOwnership.setETag("default eTag");
                     Transaction transaction = jedis.multi();
                     transaction.hset(key, PARTITION_OWNERSHIP, DEFAULT_SERIALIZER.serializeToBytes(partitionOwnership));
                     List<Object> executionResponse = transaction.exec();
 
                     if (executionResponse == null) {
                         //This means that the transaction did not execute, which implies that another client has changed the ownership during this transaction
-                        sink.error(new RuntimeException("Ownership records were changed by another client"));
+                        LOGGER.verbose("Unable to claim partition with id: " + partitionId);
+                    } else {
+                        sink.next(partitionOwnership);
+                        sink.complete();
                     }
                 }
-                jedisPool.returnResource(jedis);
             }
-            sink.next(partitionOwnership);
         }));
     }
 
@@ -96,7 +114,6 @@ public Flux<Checkpoint> listCheckpoints(String fullyQualifiedNamespace, String e
             Set<byte[]> members = jedis.smembers(prefix);
 
             if (members.isEmpty()) {
-                jedisPool.returnResource(jedis);
                 return Flux.fromIterable(listStoredCheckpoints);
             }
             for (byte[] member : members) {
@@ -116,7 +133,6 @@ public Flux<Checkpoint> listCheckpoints(String fullyQualifiedNamespace, String e
                     LOGGER.verbose("No checkpoint persists yet.");
                 }
             }
-            jedisPool.returnResource(jedis);
             return Flux.fromIterable(listStoredCheckpoints);
         }
     }
@@ -138,7 +154,6 @@ public Flux<PartitionOwnership> listOwnership(String fullyQualifiedNamespace, St
             ArrayList<PartitionOwnership> listStoredOwnerships = new ArrayList<>();
 
             if (members.isEmpty()) {
-                jedisPool.returnResource(jedis);
                 return Flux.fromIterable(listStoredOwnerships);
             }
             for (byte[] member : members) {
@@ -157,7 +172,6 @@ public Flux<PartitionOwnership> listOwnership(String fullyQualifiedNamespace, St
                     listStoredOwnerships.add(partitionOwnership);
                 }
             }
-            jedisPool.returnResource(jedis);
             return Flux.fromIterable(listStoredOwnerships);
         }
     }
@@ -188,8 +202,6 @@ public Mono<Void> updateCheckpoint(Checkpoint checkpoint) {
                     //Case 2: checkpoint already exists in Redis cache
                     jedis.hset(key, CHECKPOINT, DEFAULT_SERIALIZER.serializeToBytes(checkpoint));
                 }
-
-                jedisPool.returnResource(jedis);
             }
         });
     }

sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/src/samples/README.md

@@ -0,0 +1,44 @@
+---
+page_type: sample
+languages:
+- java
+  products:
+- azure
+- azure-event-hubs
+- azure-cache-redis
+  urlFragment: eventhubs-checkpoint-store-jedis-samples
+---
+
+# Azure Event Hubs Checkpoint Store client library samples for Java
+
+Azure Event Hubs Checkpoint Store samples are a set of self-contained Java programs that demonstrate interacting
+with Azure Event Hubs Checkpoint Store using the client library.
+
+## Key concepts
+Key concepts are explained in detail [here][sdk_readme_key_concepts].
+
+## Getting started
+Please refer to the [Getting Started][sdk_readme_getting_started] section.
+
+## Examples
+- [Create and run an instance of JedisRedisCheckpointStore][sample_jedis_client]
+- [Consume events from all Event Hub partitions][sample_event_processor]
+
+## Troubleshooting
+See [Troubleshooting][sdk_readme_troubleshooting].
+
+## Next steps
+See [Next steps][sdk_readme_next_steps].
+
+## Contributing
+
+If you would like to become an active contributor to this project please refer to our [Contribution
+Guidelines](https://github.com/Azure/azure-sdk-for-java/blob/main/CONTRIBUTING.md) for more information.
+
+<!-- Links -->
+[sample_jedis_client]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis
+[sample_event_processor]: https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis
+[sdk_readme_getting_started]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/README.md#getting-started
+[sdk_readme_key_concepts]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/README.md#key-concepts
+[sdk_readme_next_steps]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/README.md#next-steps
+[sdk_readme_troubleshooting]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-jedis/README.md#troubleshooting)