Support for additional property types. (Azure#32559)

* Add error message for missing Encoder.

* Remove unused LOCK.

* Use underlying proton-j encoders to determine size.

* Adding test for different types.

* Update CHANGELOG

* Add suppressions for rawtypes and unchecked in tests.

* Add explanations for checkpoint store.

Author: conniey

sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Features Added
 
+- Added support for setting different value types in `EventData.getProperties()`. [32518](https://github.com/Azure/azure-sdk-for-java/issues/32518)
+  
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubMessageSerializer.java

@@ -18,6 +18,11 @@
 import org.apache.qpid.proton.amqp.messaging.Data;
 import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations;
 import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
+import org.apache.qpid.proton.codec.AMQPType;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.Encoder;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.TypeEncoding;
 import org.apache.qpid.proton.message.Message;
 
 import java.time.Instant;
@@ -41,6 +46,7 @@
  * Utility class for converting {@link EventData} to {@link Message}.
  */
 class EventHubMessageSerializer implements MessageSerializer {
+    private static final Encoder ENCODER = new EncoderImpl(new DecoderImpl());
     private static final ClientLogger LOGGER = new ClientLogger(EventHubMessageSerializer.class);
     private static final Symbol LAST_ENQUEUED_SEQUENCE_NUMBER =
         Symbol.getSymbol(MANAGEMENT_RESULT_LAST_ENQUEUED_SEQUENCE_NUMBER);
@@ -348,44 +354,24 @@ private static int getPayloadSize(Message msg) {
         return 0;
     }
 
+    @SuppressWarnings({"unchecked", "rawtypes"})
     private static int sizeof(Object obj) {
-        if (obj instanceof String) {
-            return obj.toString().length() << 1;
-        }
-
-        if (obj instanceof Symbol) {
-            return ((Symbol) obj).length() << 1;
-        }
-
-        if (obj instanceof Integer) {
-            return Integer.BYTES;
-        }
-
-        if (obj instanceof Long) {
-            return Long.BYTES;
-        }
-
-        if (obj instanceof Short) {
-            return Short.BYTES;
-        }
-
-        if (obj instanceof Character) {
-            return Character.BYTES;
-        }
-
-        if (obj instanceof Float) {
-            return Float.BYTES;
+        if (obj == null) {
+            return 0;
         }
 
-        if (obj instanceof Double) {
-            return Double.BYTES;
+        final AMQPType amqpType = ENCODER.getType(obj);
+        if (amqpType == null) {
+            throw new IllegalArgumentException(String.format(Messages.ENCODING_TYPE_NOT_SUPPORTED,
+                obj.getClass()));
         }
 
-        if (obj instanceof Date) {
-            return 32;
+        final TypeEncoding encoding = amqpType.getEncoding(obj);
+        if (encoding == null) {
+            throw new IllegalArgumentException(String.format(
+                Messages.ENCODING_TYPE_NOT_SUPPORTED_ENCODER, obj.getClass()));
         }
 
-        throw new IllegalArgumentException(String.format(Messages.ENCODING_TYPE_NOT_SUPPORTED,
-            obj.getClass()));
+        return encoding.getValueSize(obj);
     }
 }

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/Messages.java

@@ -4,6 +4,7 @@
 package com.azure.messaging.eventhubs;
 
 import com.azure.core.util.CoreUtils;
+
 import java.util.Map;
 
 /**
@@ -24,6 +25,11 @@ public enum Messages {
      */
     public static final String ENCODING_TYPE_NOT_SUPPORTED = getMessage("ENCODING_TYPE_NOT_SUPPORTED");
 
+    /**
+     * Encoder is not supported message.
+     */
+    public static final String ENCODING_TYPE_NOT_SUPPORTED_ENCODER = getMessage("ENCODING_TYPE_NOT_SUPPORTED_ENCODER");
+
     /**
      * Process span scope type error message.
      */

sdk/eventhubs/azure-messaging-eventhubs/src/main/resources/eventhubs-messages.properties

@@ -12,3 +12,4 @@ ERROR_OCCURRED_IN_SUBSCRIBER_ERROR=Error occurred in subscriber.
 EXCEPTION_OCCURRED_WHILE_EMITTING=Exception occurred while emitting next received event.
 CLASS_NOT_A_SUPPORTED_TYPE=Class '%s' is not a supported deserializable type.
 ENCODING_TYPE_NOT_SUPPORTED=Encoding Type: %s is not supported.
+ENCODING_TYPE_NOT_SUPPORTED_ENCODER=Encoding Type: %s does not have an encoder.

sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/SampleCheckpointStore.java

@@ -7,11 +7,11 @@
 import com.azure.core.util.logging.ClientLogger;
 import com.azure.messaging.eventhubs.models.Checkpoint;
 import com.azure.messaging.eventhubs.models.PartitionOwnership;
-import java.util.List;
-import java.util.Locale;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -80,10 +80,18 @@ public Flux<PartitionOwnership> claimOwnership(List<PartitionOwnership> requeste
             firstEntry.getConsumerGroup(), OWNERSHIP);
 
         return Flux.fromIterable(requestedPartitionOwnerships)
-            .filter(partitionOwnership -> {
-                return !partitionOwnershipMap.containsKey(partitionOwnership.getPartitionId())
-                    || partitionOwnershipMap.get(partitionOwnership.getPartitionId()).getETag()
-                    .equals(partitionOwnership.getETag());
+            .filter(ownershipRequest -> {
+                final PartitionOwnership existing = partitionOwnershipMap.get(ownershipRequest.getPartitionId());
+
+                // There are no existing ownership records. Safe to claim.
+                if (existing == null) {
+                    return true;
+                }
+
+                // The eTag for the ownership request matches the one we have in our store.  If they did not match,
+                // it means that between the time the ownership request was being calculated and now, another thread,
+                // process, etc. updated the blob.  Consequently, we will deny this ownership request.
+                return existing.getETag().equals(ownershipRequest.getETag());
             })
             .doOnNext(partitionOwnership ->
                 LOGGER.atInfo()

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/IntegrationTestBase.java

@@ -73,7 +73,6 @@ public abstract class IntegrationTestBase extends TestBase {
 
     private static final String AZURE_EVENTHUBS_FULLY_QUALIFIED_DOMAIN_NAME = "AZURE_EVENTHUBS_FULLY_QUALIFIED_DOMAIN_NAME";
     private static final String AZURE_EVENTHUBS_EVENT_HUB_NAME = "AZURE_EVENTHUBS_EVENT_HUB_NAME";
-    private static final Object LOCK = new Object();
     private static final Configuration GLOBAL_CONFIGURATION = Configuration.getGlobalConfiguration();
 
     private static Scheduler scheduler;
@@ -341,6 +340,6 @@ protected void dispose(Closeable... closeables) {
     }
 
     private void skipIfNotRecordMode() {
-        Assumptions.assumeTrue(getTestMode() != TestMode.PLAYBACK);
+        Assumptions.assumeTrue(getTestMode() != TestMode.PLAYBACK, "Is not in RECORD/LIVE mode.");
     }
 }