diff --git a/fluent-tests/pom.xml b/fluent-tests/pom.xml
index 3f46924612..9c30d6fc1d 100644
--- a/fluent-tests/pom.xml
+++ b/fluent-tests/pom.xml
@@ -13,13 +13,6 @@
   <artifactId>fluent-tests</artifactId>
   <version>1.0.0-SNAPSHOT</version>
 
-  <repositories>
-    <repository>
-      <id>azuresdkmaven</id>
-      <url>https://azuresdkmaven.blob.core.windows.net/snapshot/m2</url>
-    </repository>
-  </repositories>
-
   <profiles>
     <profile>
       <id>local</id>
@@ -64,13 +57,13 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-management</artifactId>
-      <version>1.0.0-beta.8-SNAPSHOT</version>
+      <version>1.0.0-beta.2</version>
     </dependency>
 
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-http-netty</artifactId>
-      <version>1.5.1</version>
+      <version>1.5.2</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -85,6 +78,12 @@
       <version>5.6.0</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-simple</artifactId>
+      <version>1.7.30</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
diff --git a/fluent-tests/src/main/java/com/azure/resourcemanager/resources/fluentcore/AzureServiceClient.java b/fluent-tests/src/main/java/com/azure/resourcemanager/resources/fluentcore/AzureServiceClient.java
index 164a423032..be0af0d3ab 100644
--- a/fluent-tests/src/main/java/com/azure/resourcemanager/resources/fluentcore/AzureServiceClient.java
+++ b/fluent-tests/src/main/java/com/azure/resourcemanager/resources/fluentcore/AzureServiceClient.java
@@ -114,7 +114,7 @@ public <T, U> PollerFlux<PollResult<T>, U> getLroResultAsync(Mono<Response<Flux<
                 httpPipeline,
                 pollResultType,
                 finalResultType,
-                Duration.ofSeconds(30),
+                Duration.ofSeconds(3),
                 lroInit);
     }
 
diff --git a/fluentgen/src/main/java/com/azure/autorest/fluent/template/FluentClientMethodTemplate.java b/fluentgen/src/main/java/com/azure/autorest/fluent/template/FluentClientMethodTemplate.java
index 8b68d80f46..cbaa6be13c 100644
--- a/fluentgen/src/main/java/com/azure/autorest/fluent/template/FluentClientMethodTemplate.java
+++ b/fluentgen/src/main/java/com/azure/autorest/fluent/template/FluentClientMethodTemplate.java
@@ -28,6 +28,7 @@ public static FluentClientMethodTemplate getInstance() {
     protected void generatePagedAsyncSinglePage(ClientMethod clientMethod, JavaType typeBlock, ProxyMethod restAPIMethod, JavaSettings settings) {
         boolean addContextParameter = settings.getAddContextParameter() && !(settings.isContextClientMethodParameter() && contextInParameters(clientMethod));
         boolean mergeContextParameter = settings.getAddContextParameter() && (settings.isContextClientMethodParameter() && contextInParameters(clientMethod));
+        boolean isLroPagination = GenericType.Mono(GenericType.Response(GenericType.FluxByteBuffer)).equals(restAPIMethod.getReturnType().getClientType());
         String endOfLine = addContextParameter ? "" : ";";
 
         typeBlock.annotation("ServiceMethod(returns = ReturnType.SINGLE)");
@@ -43,11 +44,35 @@ protected void generatePagedAsyncSinglePage(ClientMethod clientMethod, JavaType
                     function.line(String.format("context = %s.mergeContext(context);", clientMethod.getClientReference()));
                 }
                 if (addContextParameter) {
-                    function.line(String.format("return FluxUtil.withContext(context -> %s)",
-                            serviceMethodCall));
+                    if (!isLroPagination) {
+                        function.line(String.format("return FluxUtil.withContext(context -> %s)",
+                                serviceMethodCall));
+                    } else {
+                        function.line("return FluxUtil.withContext(context -> {");
+                        function.indent(() -> {
+                            function.line(String.format("%s mono = %s.cache();",
+                                    clientMethod.getProxyMethod().getReturnType().toString(),
+                                    serviceMethodCall));
+
+                            IType classType = clientMethod.getMethodPageDetails().getLroIntermediateType();
+                            function.line(String.format("return Mono.zip(mono, %s.<%s, %s>getLroResultAsync(mono, %s.getHttpPipeline(), %s.class, %s.class).last().flatMap(%s::getLroFinalResultOrError));",
+                                    clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference()));
+                        });
+                        function.line("})");
+                    }
                 } else {
-                    function.line(String.format("return %s",
-                            serviceMethodCall));
+                    if (!isLroPagination) {
+                        function.line(String.format("return %s",
+                                serviceMethodCall));
+                    } else {
+                        function.line(String.format("%s mono = %s.cache();",
+                                clientMethod.getProxyMethod().getReturnType().toString(),
+                                serviceMethodCall));
+
+                        IType classType = clientMethod.getMethodPageDetails().getLroIntermediateType();
+                        function.line(String.format("return Mono.zip(mono, %s.<%s, %s>getLroResultAsync(mono, %s.getHttpPipeline(), %s.class, %s.class).last().flatMap(%s::getLroFinalResultOrError))",
+                                clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference()));
+                    }
                 }
                 function.indent(() -> {
                     if (addContextParameter) {
@@ -57,16 +82,30 @@ protected void generatePagedAsyncSinglePage(ClientMethod clientMethod, JavaType
                         function.line(".map(res -> new PagedResponseBase<>(");
                     }
                     function.indent(() -> {
-                        function.line("res.getRequest(),");
-                        function.line("res.getStatusCode(),");
-                        function.line("res.getHeaders(),");
-                        function.line("res.getValue().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getItemName()));
-                        function.line("res.getValue().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getNextLinkName()));
-                        IType responseType = ((GenericType) clientMethod.getProxyMethod().getReturnType()).getTypeArguments()[0];
-                        if (responseType instanceof ClassType) {
-                            function.line(String.format("res.getDeserializedHeaders()))%s", endOfLine));
+                        if (!isLroPagination) {
+                            function.line("res.getRequest(),");
+                            function.line("res.getStatusCode(),");
+                            function.line("res.getHeaders(),");
+                            function.line("res.getValue().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getItemName()));
+                            function.line("res.getValue().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getNextLinkName()));
+                            IType responseType = ((GenericType) clientMethod.getProxyMethod().getReturnType()).getTypeArguments()[0];
+                            if (responseType instanceof ClassType) {
+                                function.line(String.format("res.getDeserializedHeaders()))%s", endOfLine));
+                            } else {
+                                function.line(String.format("null))%s", endOfLine));
+                            }
                         } else {
-                            function.line(String.format("null))%s", endOfLine));
+                            function.line("res.getT1().getRequest(),");
+                            function.line("res.getT1().getStatusCode(),");
+                            function.line("res.getT1().getHeaders(),");
+                            function.line("res.getT2().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getItemName()));
+                            function.line("res.getT2().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getNextLinkName()));
+                            IType responseType = ((GenericType) clientMethod.getProxyMethod().getReturnType()).getTypeArguments()[0];
+                            if (responseType instanceof ClassType) {
+                                function.line(String.format("res.getT2().getDeserializedHeaders()))%s", endOfLine));
+                            } else {
+                                function.line(String.format("null))%s", endOfLine));
+                            }
                         }
                     });
                     if (addContextParameter) {
@@ -84,11 +123,35 @@ protected void generatePagedAsyncSinglePage(ClientMethod clientMethod, JavaType
                     function.line(String.format("context = %s.mergeContext(context);", clientMethod.getClientReference()));
                 }
                 if (addContextParameter) {
-                    function.line(String.format("return FluxUtil.withContext(context -> %s)",
-                            serviceMethodCall));
+                    if (!isLroPagination) {
+                        function.line(String.format("return FluxUtil.withContext(context -> %s)",
+                                serviceMethodCall));
+                    } else {
+                        function.line("return FluxUtil.withContext(context -> {");
+                        function.indent(() -> {
+                            function.line(String.format("%s mono = %s.cache();",
+                                    clientMethod.getProxyMethod().getReturnType().toString(),
+                                    serviceMethodCall));
+
+                            IType classType = clientMethod.getMethodPageDetails().getLroIntermediateType();
+                            function.line(String.format("return Mono.zip(mono, %s.<%s, %s>getLroResultAsync(mono, %s.getHttpPipeline(), %s.class, %s.class).last().flatMap(%s::getLroFinalResultOrError));",
+                                    clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference()));
+                        });
+                        function.line("})");
+                    }
                 } else {
-                    function.line(String.format("return %s",
-                            serviceMethodCall));
+                    if (!isLroPagination) {
+                        function.line(String.format("return %s",
+                                serviceMethodCall));
+                    } else {
+                        function.line(String.format("%s mono = %s.cache();",
+                                clientMethod.getProxyMethod().getReturnType().toString(),
+                                serviceMethodCall));
+
+                        IType classType = clientMethod.getMethodPageDetails().getLroIntermediateType();
+                        function.line(String.format("return Mono.zip(mono, %s.<%s, %s>getLroResultAsync(mono, %s.getHttpPipeline(), %s.class, %s.class).last().flatMap(%s::getLroFinalResultOrError))",
+                                clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference(), classType.toString(), classType.toString(), clientMethod.getClientReference()));
+                    }
                 }
                 function.indent(() -> {
                     if (addContextParameter) {
@@ -98,16 +161,30 @@ protected void generatePagedAsyncSinglePage(ClientMethod clientMethod, JavaType
                         function.line(".map(res -> new PagedResponseBase<>(");
                     }
                     function.indent(() -> {
-                        function.line("res.getRequest(),");
-                        function.line("res.getStatusCode(),");
-                        function.line("res.getHeaders(),");
-                        function.line("res.getValue().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getItemName()));
-                        function.line("null,");
-                        IType responseType = ((GenericType) clientMethod.getProxyMethod().getReturnType()).getTypeArguments()[0];
-                        if (responseType instanceof ClassType) {
-                            function.line(String.format("res.getDeserializedHeaders()))%s", endOfLine));
+                        if (!isLroPagination) {
+                            function.line("res.getRequest(),");
+                            function.line("res.getStatusCode(),");
+                            function.line("res.getHeaders(),");
+                            function.line("res.getValue().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getItemName()));
+                            function.line("null,");
+                            IType responseType = ((GenericType) clientMethod.getProxyMethod().getReturnType()).getTypeArguments()[0];
+                            if (responseType instanceof ClassType) {
+                                function.line(String.format("res.getDeserializedHeaders()))%s", endOfLine));
+                            } else {
+                                function.line(String.format("null))%s", endOfLine));
+                            }
                         } else {
-                            function.line(String.format("null))%s", endOfLine));
+                            function.line("res.getT1().getRequest(),");
+                            function.line("res.getT1().getStatusCode(),");
+                            function.line("res.getT1().getHeaders(),");
+                            function.line("res.getT2().%s(),", CodeNamer.getModelNamer().modelPropertyGetterName(clientMethod.getMethodPageDetails().getItemName()));
+                            function.line("null,");
+                            IType responseType = ((GenericType) clientMethod.getProxyMethod().getReturnType()).getTypeArguments()[0];
+                            if (responseType instanceof ClassType) {
+                                function.line(String.format("res.getT2().getDeserializedHeaders()))%s", endOfLine));
+                            } else {
+                                function.line(String.format("null))%s", endOfLine));
+                            }
                         }
                     });
                     if (addContextParameter) {
diff --git a/fluentnamer/pom.xml b/fluentnamer/pom.xml
index 6ce57fd8bd..af83033eba 100644
--- a/fluentnamer/pom.xml
+++ b/fluentnamer/pom.xml
@@ -48,7 +48,7 @@
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-simple</artifactId>
-      <version>1.7.28</version>
+      <version>1.7.30</version>
     </dependency>
   </dependencies>
 
diff --git a/fluentnamer/src/main/java/com/azure/autorest/fluent/transformer/FluentTransformer.java b/fluentnamer/src/main/java/com/azure/autorest/fluent/transformer/FluentTransformer.java
index eabbdffcee..15eedfe2b1 100644
--- a/fluentnamer/src/main/java/com/azure/autorest/fluent/transformer/FluentTransformer.java
+++ b/fluentnamer/src/main/java/com/azure/autorest/fluent/transformer/FluentTransformer.java
@@ -30,7 +30,6 @@ public FluentTransformer(FluentJavaSettings fluentJavaSettings) {
     }
 
     public CodeModel preTransform(CodeModel codeModel) {
-        codeModel = removePagingLRO(codeModel);
         if (fluentJavaSettings.getNameForUngroupedOperations().isPresent()) {
             codeModel = renameUngroupedOperationGroup(codeModel, fluentJavaSettings.getNameForUngroupedOperations().get());
         }
@@ -52,13 +51,6 @@ public CodeModel postTransform(CodeModel codeModel) {
         return codeModel;
     }
 
-    public CodeModel removePagingLRO(CodeModel codeModel) {
-        codeModel.getOperationGroups().stream().flatMap(og -> og.getOperations().stream())
-                .filter(o -> hasLongRunningOperationExtension(o) && hasPaging(o))
-                .forEach(o -> o.getExtensions().setXmsPageable(null));
-        return codeModel;
-    }
-
     public CodeModel renameUngroupedOperationGroup(CodeModel codeModel, String nameForUngroupOperations) {
         codeModel.getOperationGroups().stream()
                 .filter(og -> Utils.getDefaultName(og) == null || Utils.getDefaultName(og).isEmpty())
diff --git a/javagen/src/main/java/com/azure/autorest/mapper/ClientMethodMapper.java b/javagen/src/main/java/com/azure/autorest/mapper/ClientMethodMapper.java
index afc95345ce..1084fb4fa3 100644
--- a/javagen/src/main/java/com/azure/autorest/mapper/ClientMethodMapper.java
+++ b/javagen/src/main/java/com/azure/autorest/mapper/ClientMethodMapper.java
@@ -185,11 +185,17 @@ public List<ClientMethod> map(Operation operation) {
                 if (pageableItemName != null) {
                     boolean isNextMethod = operation.getExtensions().getXmsPageable().getNextOperation() == operation;
 
+                    IType lroIntermediateType = null;
+                    if (operation.getExtensions().isXmsLongRunningOperation() && !isNextMethod) {
+                        lroIntermediateType = SchemaUtil.getOperationResponseType(operation);
+                    }
+
                     MethodPageDetails details = new MethodPageDetails(
                             CodeNamer.getPropertyName(operation.getExtensions().getXmsPageable().getNextLinkName()),
                             pageableItemName,
                             (isNextMethod || operation.getExtensions().getXmsPageable().getNextOperation() == null) ? null : Mappers.getClientMethodMapper().map(operation.getExtensions().getXmsPageable().getNextOperation())
-                                    .stream().findFirst().get());
+                                    .stream().findFirst().get(),
+                            lroIntermediateType);
                     builder.methodPageDetails(details);
 
                     if (!(!settings.getRequiredParameterClientMethods() && settings.isContextClientMethodParameter()
diff --git a/javagen/src/main/java/com/azure/autorest/mapper/ProxyMethodMapper.java b/javagen/src/main/java/com/azure/autorest/mapper/ProxyMethodMapper.java
index eb3bbe1778..86e83ffbf2 100644
--- a/javagen/src/main/java/com/azure/autorest/mapper/ProxyMethodMapper.java
+++ b/javagen/src/main/java/com/azure/autorest/mapper/ProxyMethodMapper.java
@@ -64,7 +64,8 @@ public Map<Request, ProxyMethod> map(Operation operation) {
         IType responseBodyType = SchemaUtil.getOperationResponseType(operation);
 
         IType returnType;
-        if (operation.getExtensions() != null && operation.getExtensions().isXmsLongRunningOperation() && settings.isFluent()) {
+        if (operation.getExtensions() != null && operation.getExtensions().isXmsLongRunningOperation() && settings.isFluent()
+                && (operation.getExtensions().getXmsPageable() == null || !(operation.getExtensions().getXmsPageable().getNextOperation() == operation))) {
             returnType = GenericType.Response(GenericType.FluxByteBuffer);    // raw response for LRO
             builder.returnType(GenericType.Mono(returnType));
         } else if (operation.getResponses().stream().anyMatch(r -> Boolean.TRUE.equals(r.getBinary()))) {
diff --git a/javagen/src/main/java/com/azure/autorest/model/clientmodel/MethodPageDetails.java b/javagen/src/main/java/com/azure/autorest/model/clientmodel/MethodPageDetails.java
index 348b1ce2df..6bde77457d 100644
--- a/javagen/src/main/java/com/azure/autorest/model/clientmodel/MethodPageDetails.java
+++ b/javagen/src/main/java/com/azure/autorest/model/clientmodel/MethodPageDetails.java
@@ -15,10 +15,15 @@ public class MethodPageDetails {
     private String itemName;
     private ClientMethod nextMethod;
 
-    public MethodPageDetails(String nextLinkName, String itemName, ClientMethod nextMethod) {
+    // Proxy method return type is Flux<ByteBuffer>. Client method return type is PagedResponse<>.
+    // This intermediate type is the type of pagination response (the type with values and nextLink).
+    private IType lroIntermediateType;
+
+    public MethodPageDetails(String nextLinkName, String itemName, ClientMethod nextMethod, IType lroIntermediateType) {
         this.nextLinkName = nextLinkName;
         this.itemName = itemName;
         this.nextMethod = nextMethod;
+        this.lroIntermediateType = lroIntermediateType;
     }
 
     public String getNextLinkName() {
@@ -29,10 +34,14 @@ public String getItemName() {
         return itemName;
     }
 
-    public final ClientMethod getNextMethod() {
+    public ClientMethod getNextMethod() {
         return nextMethod;
     }
 
+    public IType getLroIntermediateType() {
+        return lroIntermediateType;
+    }
+
     public boolean nonNullNextLink() {
         return nextLinkName != null && !nextLinkName.isEmpty();
     }
diff --git a/javagen/src/main/java/com/azure/autorest/template/ClientMethodTemplate.java b/javagen/src/main/java/com/azure/autorest/template/ClientMethodTemplate.java
index 16b216fc99..2a10f3a52c 100644
--- a/javagen/src/main/java/com/azure/autorest/template/ClientMethodTemplate.java
+++ b/javagen/src/main/java/com/azure/autorest/template/ClientMethodTemplate.java
@@ -385,7 +385,7 @@ public final void write(ClientMethod clientMethod, JavaType typeBlock) {
                 typeBlock.publicMethod(clientMethod.getDeclaration(), function -> {
                     AddOptionalVariables(function, clientMethod, restAPIMethod.getParameters(), settings);
                     if (clientMethod.getReturnValue().getType() == ClassType.InputStream) {
-                        function.line("Iterator<ByteBufferBackedInputStream> iterator = %s(%S).map(ByteBufferBackedInputStream::new).toStream().iterator();",
+                        function.line("Iterator<ByteBufferBackedInputStream> iterator = %s(%s).map(ByteBufferBackedInputStream::new).toStream().iterator();",
                                 clientMethod.getSimpleAsyncMethodName(), clientMethod.getArgumentList());
                         function.anonymousClass("Enumeration<InputStream>", "enumeration", javaBlock -> {
                             javaBlock.annotation("Override");