[Storage] Queue message encoding #19328

kasobol-msft · 2021-02-19T18:55:17Z

Resolves #17460

This is a port from .NET SDK see:
Changelog
QueueMessageEncoding
QueueClientOptions.MessageEncoding
Badly encoded message handler - Problem statement
Badly encoded message handler - .NET implementation
PR1
PR2

check-enforcer · 2021-02-19T20:36:33Z

This pull request is protected by Check Enforcer.

What is Check Enforcer?

Check Enforcer helps ensure all pull requests are covered by at least one check-run (typically an Azure Pipeline). When all check-runs associated with this pull request pass then Check Enforcer itself will pass.

Why am I getting this message?

You are getting this message because Check Enforcer did not detect any check-runs being associated with this pull request within five minutes. This may indicate that your pull request is not covered by any pipelines and so Check Enforcer is correctly blocking the pull request being merged.

What should I do now?

If the check-enforcer check-run is not passing and all other check-runs associated with this PR are passing (excluding license-cla) then you could try telling Check Enforcer to evaluate your pull request again. You can do this by adding a comment to this pull request as follows:
/check-enforcer evaluate
Typically evaulation only takes a few seconds. If you know that your pull request is not covered by a pipeline and this is expected you can override Check Enforcer using the following command:
/check-enforcer override
Note that using the override command triggers alerts so that follow-up investigations can occur (PRs still need to be approved as normal).

What if I am onboarding a new service?

Often, new services do not have validation pipelines associated with them, in order to bootstrap pipelines for a new service, you can issue the following command as a pull request comment:
/azp run prepare-pipelines
This will run a pipeline that analyzes the source tree and creates the pipelines necessary to build and validate your pull request. Once the pipeline has been created you can trigger the pipeline using the following comment:
/azp run java - [service] - ci

alzimmermsft · 2021-02-23T17:40:48Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

+        } catch (RuntimeException ex) {
+            return monoError(logger, ex);
+        }


nit: This can be removed as the method being called into handles this

alzimmermsft · 2021-02-23T18:04:51Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

@@ -586,26 +638,81 @@ public HttpPipeline getHttpPipeline() {
    public Mono<Response<SendMessageResult>> sendMessageWithResponse(String messageText, Duration visibilityTimeout,
                                                                   Duration timeToLive) {
        try {
-            return withContext(context -> sendMessageWithResponse(messageText, visibilityTimeout, timeToLive, context));
+            BinaryData message = messageText == null ? null : BinaryData.fromString(messageText);


When messageText is empty do we want null BinaryData or an empty BinaryData? BinaryData.fromString is null safe and will return an empty BinaryData

This should be fine. In tests I discovered that previously passing "null" resulted in "XML format exception or so" so producing empty binarydata should be fine.

Other thing is that when message ends up being enqueued on the service side then worst case is service returning empty string at peek/receive.

alzimmermsft · 2021-02-23T18:07:43Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

                null, null,
                context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(response -> new SimpleResponse<>(response, response.getValue().get(0)));
    }

+    private String encodeMessage(BinaryData message) {


Could we turn this into a Mono<String> and have it return Mono.error on an invalid input? This will prevent introducing another scenario where we throw instead of return an error in a reactive stream (reducing the likelihood of the try/catch blocks being used in caller APIs)

alzimmermsft · 2021-02-23T18:10:45Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

+                                null)
+                        ));
+                    } else {
+                        throw logger.logExceptionAsError(e);


nit: return FluxUtil.monoError(logger, e)

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

alzimmermsft · 2021-02-23T18:16:32Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

+        List<QueueMessageItemInternal> queueMessageInternalItems = response.getValue();
+        List<QueueMessageItem> queueMessageItems = Collections.emptyList();
+        Mono<Void> mono = Mono.empty();
+        if (queueMessageInternalItems != null) {


If we inverted this to queueMessageInternalItems == null -> return empty paged response, I think we could streamline the conversion logic to use a fully reactive structure.

return Flux.fromIterable(queueMessageInternalItems) .map(/* conversion logic */) .collectList() .map(/* create paged response */);

alzimmermsft · 2021-02-23T18:16:58Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

+                    queueMessageItems.add(transformQueueMessageItemInternal(queueMessageItemInternal, messageEncoding));
+                } catch (IllegalArgumentException e) {
+                    if (messageDecodingFailedHandler != null) {
+                        mono = mono.then(messageDecodingFailedHandler.apply(


Does this failed handler return an error or just log a message?

it's customer provided so we expect that error handling is implemented inside that handler. However if they get it wrong we want to surface errors in receive/peek API call result (like we do in .NET).
See Peek with handler exception test.

alzimmermsft · 2021-02-23T18:18:56Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueClientBuilder.java

+     */
+    public QueueClientBuilder messageDecodingFailedHandler(
+        Function<QueueMessageDecodingFailure, Mono<Void>> messageDecodingFailedHandler) {
+        this.messageDecodingFailedHandler = Objects.requireNonNull(messageDecodingFailedHandler,


I noticed in the client code this was null checked and by default this is null in the builder, given that should null be an allowed value?

alzimmermsft · 2021-02-23T18:22:10Z

...rage/azure-storage-queue/src/main/java/com/azure/storage/queue/models/PeekedMessageItem.java

@@ -41,8 +42,7 @@
    /*
     * The content of the Message.
     */
-    @JsonProperty(value = "MessageText", required = true)
-    private String messageText;
+    private BinaryData body;


Is the serializer/deserialization logic going to handle this correctly?

PeekedMessageItemInternal and QueueMessageItemInteral has been created to handle that part.
The data flow is:
receive from wire -> from XML ->PeekedMessageItemInternal -> none/base64 decoding -> PeekedMessageItem

srnagar · 2021-02-23T19:35:04Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

+     * unset the value will default to 0 and the message will be instantly visible. The timeout must be between 0
+     * seconds and 7 days.
+     * @param timeToLive Optional. How long the message will stay alive in the queue. If unset the value will default to
+     * 7 days, if -1 is passed the message will not expire. The time to live must be -1 or any positive number.


Since timeToLive is of type Duration, setting -1 is not an option. It might be better to make it clear how to set this - Duration.ofDays(-1)

actually it's -1 seconds.

srnagar · 2021-02-23T19:42:51Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueClientBuilder.java

+     * @return the updated QueueClientBuilder object
+     * @throws NullPointerException If {@code messageDecodingFailedHandler} is {@code null}.
+     */
+    public QueueClientBuilder messageDecodingFailedHandler(


We have used process* for handlers. So, here we would name this as processMessageDecodingError(). Also, does this need to be a Function<>? Can this instead be a Consumer<QueueMessageDecodingFailure>?

discussed offline, will provide both sync and async handler.

srnagar · 2021-02-23T20:05:38Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueMessageEncoding.java

+/**
+ * Determines how queue message body is represented in HTTP requests and responses.
+ */
+public enum QueueMessageEncoding {


Do we plan on adding more encoding formats? If so, this should be ExpandableStringEnum.

Unlikely. Besides per https://azure.github.io/azure-sdk/java_introduction.html#enumerations I don't think this is a case for ExpandableStringEnum . The message encoding is client side concept and we don't expect neither service nor user to provide values that are unknown at given SDK version. Therefore I think normal enum is good fit.

srnagar · 2021-02-23T20:57:03Z

...age/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceClientBuilder.java

+     */
+    public QueueServiceClientBuilder messageDecodingFailedHandler(
+        Function<QueueMessageDecodingFailure, Mono<Void>> messageDecodingFailedHandler) {
+        this.messageDecodingFailedHandler = Objects.requireNonNull(messageDecodingFailedHandler,


Similar to Alan's comment above, this should allow setting null.

srnagar · 2021-02-23T20:57:58Z

...age/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceClientBuilder.java

+     * @throws NullPointerException If {@code messageDecodingFailedHandler} is {@code null}.
+     */
+    public QueueServiceClientBuilder messageDecodingFailedHandler(
+        Function<QueueMessageDecodingFailure, Mono<Void>> messageDecodingFailedHandler) {


Consumer<QueueMessageDecodingFailure>

srnagar · 2021-02-23T20:58:31Z

...age/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceClientBuilder.java

+     * @return the updated QueueServiceClientBuilder object
+     * @throws NullPointerException If {@code messageDecodingFailedHandler} is {@code null}.
+     */
+    public QueueServiceClientBuilder messageDecodingFailedHandler(


processMessageDecodingError

srnagar · 2021-02-23T21:00:34Z

...-storage-queue/src/main/java/com/azure/storage/queue/models/QueueMessageDecodingFailure.java

+ * Contains information about message that could not be decoded.
+ */
+@Immutable
+public class QueueMessageDecodingFailure {


Make this final. Also, consider naming this type as QueueMessageDecodingError.

srnagar · 2021-02-23T21:03:43Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

+                } catch (IllegalArgumentException e) {
+                    if (messageDecodingFailedHandler != null) {
+                        mono = mono.then(messageDecodingFailedHandler.apply(
+                            new QueueMessageDecodingFailure(
+                                this,
+                                null,
+                                transformPeekedMessageItemInternal(
+                                    peekedMessageItemInternal, QueueMessageEncoding.NONE))
+                        ));
+                    } else {


When calling the user's error handler, we should include the exception information that was thrown when decoding.

srnagar · 2021-02-23T21:04:32Z

...-storage-queue/src/main/java/com/azure/storage/queue/models/QueueMessageDecodingFailure.java

+    public QueueMessageDecodingFailure(
+        QueueAsyncClient queueAsyncClient,
+        QueueMessageItem queueMessageItem,
+        PeekedMessageItem peekedMessageItem) {


Consider including the exception that was thrown when decoding here, so the customer can actually take action based on the exception.

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

srnagar · 2021-02-25T00:33:06Z

sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java

+            peekedMessageInternalItems = Collections.emptyList();
+        }
+        return Flux.fromIterable(peekedMessageInternalItems)
+            .flatMap(peekedMessageItemInternal ->


flatMap can potentially mess up the order of peeked messages. If the order is important, use flatMapSequential or concatMap instead.

kasobol-msft added 2 commits February 19, 2021 09:41

settings.

6d24c99

internal model.

16d701f

ghost added the Storage Storage Service (Queues, Blobs, Files) label Feb 19, 2021

kasobol-msft added 2 commits February 19, 2021 11:37

wip.

be03051

decoding.

1444ce8

kasobol-msft added 14 commits February 19, 2021 13:31

encode message.

81e762f

enqueue/dequeue tests.

7bdd8cb

recording.

6fefc92

undo models.

4c31940

more tests.

ab6ce4f

handler tests.

b5b85d9

bundle failure args.

d50f144

checkstyle.

8ebe4c4

checkstyle.

a4564a2

readme.

085f9ab

expose queue client.

87440c2

handler sample.

e3762ef

CHANGELOG.md

2ccec4d

Merge remote-tracking branch 'upstream/master' into queues-encoding

8756948

kasobol-msft marked this pull request as ready for review February 22, 2021 22:51

kasobol-msft requested review from alzimmermsft, amishra-dev, gapra-msft, jaschrep-msft and rickle-msft as code owners February 22, 2021 22:51

kasobol-msft added 2 commits February 22, 2021 15:25

Merge remote-tracking branch 'upstream/master' into queues-encoding

01f0a4e

fix npe.

d5fa36b

rickle-msft approved these changes Feb 23, 2021

View reviewed changes

alzimmermsft reviewed Feb 23, 2021

View reviewed changes

srnagar reviewed Feb 23, 2021

View reviewed changes

kasobol-msft added 4 commits February 23, 2021 14:00

some of pr feedback.

ca591ad

one more test.

19b52ca

two handlers.

4ee43ed

pr feedback.

7ed5124

alzimmermsft reviewed Feb 24, 2021

View reviewed changes

srnagar approved these changes Feb 25, 2021

View reviewed changes

pr feedback.

5640f8c

alzimmermsft approved these changes Feb 25, 2021

View reviewed changes

kasobol-msft merged commit cb09813 into Azure:master Feb 25, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Storage] Queue message encoding #19328

[Storage] Queue message encoding #19328

kasobol-msft commented Feb 19, 2021 •

edited

Loading

check-enforcer bot commented Feb 19, 2021

alzimmermsft Feb 23, 2021

alzimmermsft Feb 23, 2021

kasobol-msft Feb 23, 2021

kasobol-msft Feb 23, 2021

alzimmermsft Feb 23, 2021

alzimmermsft Feb 23, 2021

alzimmermsft Feb 23, 2021

alzimmermsft Feb 23, 2021

kasobol-msft Feb 23, 2021

alzimmermsft Feb 23, 2021

alzimmermsft Feb 23, 2021

kasobol-msft Feb 23, 2021

srnagar Feb 23, 2021

kasobol-msft Feb 23, 2021

srnagar Feb 23, 2021

kasobol-msft Feb 23, 2021

srnagar Feb 23, 2021

kasobol-msft Feb 23, 2021

srnagar Feb 23, 2021

srnagar Feb 23, 2021

srnagar Feb 23, 2021

srnagar Feb 23, 2021

srnagar Feb 23, 2021

srnagar Feb 23, 2021

srnagar Feb 25, 2021

[Storage] Queue message encoding #19328

[Storage] Queue message encoding #19328

Conversation

kasobol-msft commented Feb 19, 2021 • edited Loading

check-enforcer bot commented Feb 19, 2021

What is Check Enforcer?

Why am I getting this message?

What should I do now?

What if I am onboarding a new service?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kasobol-msft commented Feb 19, 2021 •

edited

Loading