Fix marshalling and Netty async bugs

 - Fix hasPayloadMembers() to ensure that the shape is an event before checking
   for unbound event members. Otherwise, the test will always return true for
   any shape that has member.

 - For non full duplex requests, continue to perform a channel.read() call only
   after writing the request to avoid ReadTimeoutExceptions.

Author: dagnir

codegen/src/main/java/software/amazon/awssdk/codegen/model/intermediate/ShapeModel.java

@@ -196,8 +196,8 @@ public List<MemberModel> getUnboundEventMembers() {
     public boolean hasPayloadMembers() {
         return hasPayloadMember ||
                getExplicitEventPayloadMember() != null ||
-               getUnboundMembers().size() > 0 ||
-               getUnboundEventMembers().size() > 0;
+               !getUnboundMembers().isEmpty() ||
+               (isEvent() && !getUnboundEventMembers().isEmpty());
     }
 
     /**

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NettyRequestExecutor.java

@@ -164,24 +164,25 @@ private void writeRequest(HttpRequest request) {
                .addListener(wireCall -> {
                    // Done writing so remove the idle write timeout handler
                    ChannelUtils.removeIfExists(channel.pipeline(), WriteTimeoutHandler.class);
-                   if (wireCall.isSuccess()) {
+                   if (wireCall.isSuccess() && !context.executeRequest().fullDuplex()) {
                        // Starting read so add the idle read timeout handler, removed when channel is released
                        channel.pipeline().addFirst(new ReadTimeoutHandler(context.configuration().readTimeoutMillis(),
                                                                           TimeUnit.MILLISECONDS));
+                       channel.read();
+
                    } else {
                        // TODO: Are there cases where we can keep the channel open?
                        closeAndRelease(channel);
                        handleFailure(() -> "Failed to make request to " + endpoint(), wireCall.cause());
                    }
                });
 
+        // FullDuplex calls need to start reading at the same time we make the request.
         if (context.executeRequest().fullDuplex()) {
             channel.pipeline().addFirst(new ReadTimeoutHandler(context.configuration().readTimeoutMillis(),
                                                                TimeUnit.MILLISECONDS));
+            channel.read();
         }
-
-        // Auto-read is turned off so trigger an explicit read to give control to HttpStreamsClientHandler
-        channel.read();
     }
 
     private URI endpoint() {