[KYUUBI #1948] Upgrade thrift version to 0.16.0

dev/dependencyList

@@ -36,6 +36,8 @@ hk2-api/2.6.1//hk2-api-2.6.1.jar
 hk2-locator/2.6.1//hk2-locator-2.6.1.jar
 hk2-utils/2.6.1//hk2-utils-2.6.1.jar
 htrace-core4/4.1.0-incubating//htrace-core4-4.1.0-incubating.jar
+httpclient/4.5.13//httpclient-4.5.13.jar
+httpcore/4.4.15//httpcore-4.4.15.jar
 jackson-annotations/2.13.1//jackson-annotations-2.13.1.jar
 jackson-core/2.13.1//jackson-core-2.13.1.jar
 jackson-databind/2.13.1//jackson-databind-2.13.1.jar
@@ -70,7 +72,7 @@ jetty-util-ajax/9.4.41.v20210516//jetty-util-ajax-9.4.41.v20210516.jar
 jetty-util/9.4.41.v20210516//jetty-util-9.4.41.v20210516.jar
 jline/0.9.94//jline-0.9.94.jar
 libfb303/0.9.3//libfb303-0.9.3.jar
-libthrift/0.9.3//libthrift-0.9.3.jar
+libthrift/0.16.0//libthrift-0.16.0.jar
 log4j-1.2-api/2.17.1//log4j-1.2-api-2.17.1.jar
 log4j-api/2.17.1//log4j-api-2.17.1.jar
 log4j-core/2.17.1//log4j-core-2.17.1.jar

dev/kyuubi-extension-spark-3-1/src/test/java/org/apache/thrift/transport/TFramedTransport.java

@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.thrift.transport;
+
+import org.apache.thrift.TByteArrayOutputStream;
+import org.apache.thrift.TConfiguration;
+
+/**
+ * This is based on libthrift-0.12.0 {@link TFramedTransport}. To fix class of
+ * org.apache.thrift.transport.TFramedTransport not found after upgrading libthrift.
+ *
+ * <p>TFramedTransport is a buffered TTransport that ensures a fully read message every time by
+ * preceding messages with a 4-byte frame size.
+ */
+public class TFramedTransport extends TTransport {
+
+  protected static final int DEFAULT_MAX_LENGTH = 16384000;
+
+  private int maxLength_;
+
+  /** Underlying transport */
+  private TTransport transport_ = null;
+
+  /** Buffer for output */
+  private final TByteArrayOutputStream writeBuffer_ = new TByteArrayOutputStream(1024);
+
+  /** Buffer for input */
+  private final TMemoryInputTransport readBuffer_ = new TMemoryInputTransport(new byte[0]);
+
+  public static class Factory extends TTransportFactory {
+    private int maxLength_;
+
+    public Factory() {
+      maxLength_ = TFramedTransport.DEFAULT_MAX_LENGTH;
+    }
+
+    public Factory(int maxLength) {
+      maxLength_ = maxLength;
+    }
+
+    @Override
+    public TTransport getTransport(TTransport base) throws TTransportException {
+      return new TFramedTransport(base, maxLength_);
+    }
+  }
+
+  /** Constructor wraps around another transport */
+  public TFramedTransport(TTransport transport, int maxLength) throws TTransportException {
+    transport_ = transport;
+    maxLength_ = maxLength;
+  }
+
+  public TFramedTransport(TTransport transport) throws TTransportException {
+    transport_ = transport;
+    maxLength_ = TFramedTransport.DEFAULT_MAX_LENGTH;
+  }
+
+  public void open() throws TTransportException {
+    transport_.open();
+  }
+
+  public boolean isOpen() {
+    return transport_.isOpen();
+  }
+
+  public void close() {
+    transport_.close();
+  }
+
+  public int read(byte[] buf, int off, int len) throws TTransportException {
+    int got = readBuffer_.read(buf, off, len);
+    if (got > 0) {
+      return got;
+    }
+
+    // Read another frame of data
+    readFrame();
+
+    return readBuffer_.read(buf, off, len);
+  }
+
+  @Override
+  public byte[] getBuffer() {
+    return readBuffer_.getBuffer();
+  }
+
+  @Override
+  public int getBufferPosition() {
+    return readBuffer_.getBufferPosition();
+  }
+
+  @Override
+  public int getBytesRemainingInBuffer() {
+    return readBuffer_.getBytesRemainingInBuffer();
+  }
+
+  @Override
+  public void consumeBuffer(int len) {
+    readBuffer_.consumeBuffer(len);
+  }
+
+  @Override
+  public TConfiguration getConfiguration() {
+    return null;
+  }
+
+  @Override
+  public void updateKnownMessageSize(long l) throws TTransportException {}
+
+  @Override
+  public void checkReadBytesAvailable(long l) throws TTransportException {}
+
+  public void clear() {
+    readBuffer_.clear();
+  }
+
+  private final byte[] i32buf = new byte[4];
+
+  private void readFrame() throws TTransportException {
+    transport_.readAll(i32buf, 0, 4);
+    int size = decodeFrameSize(i32buf);
+
+    if (size < 0) {
+      close();
+      throw new TTransportException(
+          TTransportException.CORRUPTED_DATA, "Read a negative frame size (" + size + ")!");
+    }
+
+    if (size > maxLength_) {
+      close();
+      throw new TTransportException(
+          TTransportException.CORRUPTED_DATA,
+          "Frame size (" + size + ") larger than max length (" + maxLength_ + ")!");
+    }
+
+    byte[] buff = new byte[size];
+    transport_.readAll(buff, 0, size);
+    readBuffer_.reset(buff);
+  }
+
+  public void write(byte[] buf, int off, int len) throws TTransportException {
+    writeBuffer_.write(buf, off, len);
+  }
+
+  @Override
+  public void flush() throws TTransportException {
+    byte[] buf = writeBuffer_.get();
+    int len = writeBuffer_.len();
+    writeBuffer_.reset();
+
+    encodeFrameSize(len, i32buf);
+    transport_.write(i32buf, 0, 4);
+    transport_.write(buf, 0, len);
+    transport_.flush();
+  }
+
+  public static final void encodeFrameSize(final int frameSize, final byte[] buf) {
+    buf[0] = (byte) (0xff & (frameSize >> 24));
+    buf[1] = (byte) (0xff & (frameSize >> 16));
+    buf[2] = (byte) (0xff & (frameSize >> 8));
+    buf[3] = (byte) (0xff & (frameSize));
+  }
+
+  public static final int decodeFrameSize(final byte[] buf) {
+    return ((buf[0] & 0xff) << 24)
+        | ((buf[1] & 0xff) << 16)
+        | ((buf[2] & 0xff) << 8)
+        | ((buf[3] & 0xff));
+  }
+}

dev/kyuubi-extension-spark-3-2/src/test/java/org/apache/thrift/transport/TFramedTransport.java

@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.thrift.transport;
+
+import org.apache.thrift.TByteArrayOutputStream;
+import org.apache.thrift.TConfiguration;
+
+/**
+ * This is based on libthrift-0.12.0 {@link TFramedTransport}. To fix class of
+ * org.apache.thrift.transport.TFramedTransport not found after upgrading libthrift.
+ *
+ * <p>TFramedTransport is a buffered TTransport that ensures a fully read message every time by
+ * preceding messages with a 4-byte frame size.
+ */
+public class TFramedTransport extends TTransport {
+
+  protected static final int DEFAULT_MAX_LENGTH = 16384000;
+
+  private int maxLength_;
+
+  /** Underlying transport */
+  private TTransport transport_ = null;
+
+  /** Buffer for output */
+  private final TByteArrayOutputStream writeBuffer_ = new TByteArrayOutputStream(1024);
+
+  /** Buffer for input */
+  private final TMemoryInputTransport readBuffer_ = new TMemoryInputTransport(new byte[0]);
+
+  public static class Factory extends TTransportFactory {
+    private int maxLength_;
+
+    public Factory() {
+      maxLength_ = TFramedTransport.DEFAULT_MAX_LENGTH;
+    }
+
+    public Factory(int maxLength) {
+      maxLength_ = maxLength;
+    }
+
+    @Override
+    public TTransport getTransport(TTransport base) throws TTransportException {
+      return new TFramedTransport(base, maxLength_);
+    }
+  }
+
+  /** Constructor wraps around another transport */
+  public TFramedTransport(TTransport transport, int maxLength) throws TTransportException {
+    transport_ = transport;
+    maxLength_ = maxLength;
+  }
+
+  public TFramedTransport(TTransport transport) throws TTransportException {
+    transport_ = transport;
+    maxLength_ = TFramedTransport.DEFAULT_MAX_LENGTH;
+  }
+
+  public void open() throws TTransportException {
+    transport_.open();
+  }
+
+  public boolean isOpen() {
+    return transport_.isOpen();
+  }
+
+  public void close() {
+    transport_.close();
+  }
+
+  public int read(byte[] buf, int off, int len) throws TTransportException {
+    int got = readBuffer_.read(buf, off, len);
+    if (got > 0) {
+      return got;
+    }
+
+    // Read another frame of data
+    readFrame();
+
+    return readBuffer_.read(buf, off, len);
+  }
+
+  @Override
+  public byte[] getBuffer() {
+    return readBuffer_.getBuffer();
+  }
+
+  @Override
+  public int getBufferPosition() {
+    return readBuffer_.getBufferPosition();
+  }
+
+  @Override
+  public int getBytesRemainingInBuffer() {
+    return readBuffer_.getBytesRemainingInBuffer();
+  }
+
+  @Override
+  public void consumeBuffer(int len) {
+    readBuffer_.consumeBuffer(len);
+  }
+
+  @Override
+  public TConfiguration getConfiguration() {
+    return null;
+  }
+
+  @Override
+  public void updateKnownMessageSize(long l) throws TTransportException {}
+
+  @Override
+  public void checkReadBytesAvailable(long l) throws TTransportException {}
+
+  public void clear() {
+    readBuffer_.clear();
+  }
+
+  private final byte[] i32buf = new byte[4];
+
+  private void readFrame() throws TTransportException {
+    transport_.readAll(i32buf, 0, 4);
+    int size = decodeFrameSize(i32buf);
+
+    if (size < 0) {
+      close();
+      throw new TTransportException(
+          TTransportException.CORRUPTED_DATA, "Read a negative frame size (" + size + ")!");
+    }
+
+    if (size > maxLength_) {
+      close();
+      throw new TTransportException(
+          TTransportException.CORRUPTED_DATA,
+          "Frame size (" + size + ") larger than max length (" + maxLength_ + ")!");
+    }
+
+    byte[] buff = new byte[size];
+    transport_.readAll(buff, 0, size);
+    readBuffer_.reset(buff);
+  }
+
+  public void write(byte[] buf, int off, int len) throws TTransportException {
+    writeBuffer_.write(buf, off, len);
+  }
+
+  @Override
+  public void flush() throws TTransportException {
+    byte[] buf = writeBuffer_.get();
+    int len = writeBuffer_.len();
+    writeBuffer_.reset();
+
+    encodeFrameSize(len, i32buf);
+    transport_.write(i32buf, 0, 4);
+    transport_.write(buf, 0, len);
+    transport_.flush();
+  }
+
+  public static final void encodeFrameSize(final int frameSize, final byte[] buf) {
+    buf[0] = (byte) (0xff & (frameSize >> 24));
+    buf[1] = (byte) (0xff & (frameSize >> 16));
+    buf[2] = (byte) (0xff & (frameSize >> 8));
+    buf[3] = (byte) (0xff & (frameSize));
+  }
+
+  public static final int decodeFrameSize(final byte[] buf) {
+    return ((buf[0] & 0xff) << 24)
+        | ((buf[1] & 0xff) << 16)
+        | ((buf[2] & 0xff) << 8)
+        | ((buf[3] & 0xff));
+  }
+}