fix review

apache · Feb 18, 2024 · 374711e · 374711e
1 parent c94141b
commit 374711e
Show file tree

Hide file tree

Showing 8 changed files with 23 additions and 17 deletions.
diff --git a/...a/org/apache/paimon/StreamLoadServer.java → ...ache/paimon/gateway/StreamLoadServer.java b/...a/org/apache/paimon/StreamLoadServer.java → ...ache/paimon/gateway/StreamLoadServer.java
@@ -16,15 +16,15 @@
  * limitations under the License.
  */
 
-package org.apache.paimon;
+package org.apache.paimon.gateway;
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.CatalogFactory;
-import org.apache.paimon.config.NettyServerConfig;
-import org.apache.paimon.exception.RemoteException;
+import org.apache.paimon.gateway.config.NettyServerConfig;
+import org.apache.paimon.gateway.exception.RemoteException;
 import org.apache.paimon.options.Options;
-import org.apache.paimon.handler.LoadHttpHandler;
+import org.apache.paimon.gateway.handler.LoadHttpHandler;
 
 import org.apache.paimon.shade.guava30.com.google.common.util.concurrent.ThreadFactoryBuilder;
 
@@ -40,17 +40,14 @@
 import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup;
 import org.apache.flink.shaded.netty4.io.netty.channel.socket.SocketChannel;
 import org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioServerSocketChannel;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpContentDecompressor;
 import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpServerCodec;
 import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedWriteHandler;
-import org.apache.flink.shaded.netty4.io.netty.handler.timeout.IdleStateHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.apache.flink.shaded.netty4.io.netty.channel.epoll.Epoll.isAvailable;

diff --git a/...ache/paimon/config/NettyServerConfig.java → ...mon/gateway/config/NettyServerConfig.java b/...ache/paimon/config/NettyServerConfig.java → ...mon/gateway/config/NettyServerConfig.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.config;
+package org.apache.paimon.gateway.config;
 
 /** dfdf. */
 public class NettyServerConfig {

diff --git a/...che/paimon/exception/RemoteException.java → ...on/gateway/exception/RemoteException.java b/...che/paimon/exception/RemoteException.java → ...on/gateway/exception/RemoteException.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.exception;
+package org.apache.paimon.gateway.exception;
 
 /** Custom runtime exception. */
 public class RemoteException extends RuntimeException {

diff --git a/...pache/paimon/handler/LoadHttpHandler.java → ...imon/gateway/handler/LoadHttpHandler.java b/...pache/paimon/handler/LoadHttpHandler.java → ...imon/gateway/handler/LoadHttpHandler.java
@@ -16,12 +16,16 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.handler;
+package org.apache.paimon.gateway.handler;
 
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultFullHttpResponse;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.LastHttpContent;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.reader.ExcelWriteStrategy;
-import org.apache.paimon.reader.WriteStrategy;
+import org.apache.paimon.gateway.reader.ExcelWriteStrategy;
+import org.apache.paimon.gateway.reader.WriteStrategy;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.sink.BatchTableCommit;
@@ -49,7 +53,7 @@
 import java.util.List;
 import java.util.Optional;
 
-import static org.apache.paimon.utils.HttpServerResponseUtil.response;
+import static org.apache.paimon.gateway.utils.HttpServerResponseUtil.response;
 
 /** An HTTP handler for loading data into a catalog, handling various HTTP events and requests. */
 @ChannelHandler.Sharable
@@ -174,6 +178,11 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Ex
                             + "    \"CommitAndPublishTimeMs\": 106\n"
                             + "}");
         }
+        if (msg instanceof LastHttpContent) {
+            // 所有的数据块都已经接收完毕
+            ctx.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK));
+            // 这里可以清理资源，比如关闭文件输出流等
+        }
     }
 
     private static Optional<String[]> extractDbAndTable(String path) {

diff --git a/...che/paimon/reader/ExcelWriteStrategy.java → ...on/gateway/reader/ExcelWriteStrategy.java b/...che/paimon/reader/ExcelWriteStrategy.java → ...on/gateway/reader/ExcelWriteStrategy.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.reader;
+package org.apache.paimon.gateway.reader;
 
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.table.sink.BatchTableWrite;

diff --git a/...g/apache/paimon/reader/WriteStrategy.java → .../paimon/gateway/reader/WriteStrategy.java b/...g/apache/paimon/reader/WriteStrategy.java → .../paimon/gateway/reader/WriteStrategy.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.reader;
+package org.apache.paimon.gateway.reader;
 
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.table.sink.BatchTableWrite;

diff --git a/.../paimon/utils/HttpServerResponseUtil.java → ...gateway/utils/HttpServerResponseUtil.java b/.../paimon/utils/HttpServerResponseUtil.java → ...gateway/utils/HttpServerResponseUtil.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.utils;
+package org.apache.paimon.gateway.utils;
 
 import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
 import org.apache.flink.shaded.netty4.io.netty.buffer.Unpooled;

diff --git a/...he/paimon/utils/RequestParameterUtil.java → ...n/gateway/utils/RequestParameterUtil.java b/...he/paimon/utils/RequestParameterUtil.java → ...n/gateway/utils/RequestParameterUtil.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.utils;
+package org.apache.paimon.gateway.utils;
 
 import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
 import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.QueryStringDecoder;