ODP-780 Update libthrift from 0.12.0 to 0.14.1 for Hive 3.1.4 connection

dev/deps/spark-deps-hadoop-2.7-hive-2.3

@@ -178,7 +178,7 @@ kubernetes-model-storageclass/5.4.1//kubernetes-model-storageclass-5.4.1.jar
 lapack/2.2.1//lapack-2.2.1.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 libfb303/0.9.3//libfb303-0.9.3.jar
-libthrift/0.12.0//libthrift-0.12.0.jar
+libthrift/0.14.1//libthrift-0.12.0.jar
 log4j/1.2.17//log4j-1.2.17.jar
 logging-interceptor/3.12.12//logging-interceptor-3.12.12.jar
 lz4-java/1.7.1//lz4-java-1.7.1.jar

dev/deps/spark-deps-hadoop-3.2-hive-2.3

@@ -149,7 +149,7 @@ kubernetes-model-storageclass/5.4.1//kubernetes-model-storageclass-5.4.1.jar
 lapack/2.2.1//lapack-2.2.1.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 libfb303/0.9.3//libfb303-0.9.3.jar
-libthrift/0.12.0//libthrift-0.12.0.jar
+libthrift/0.14.1//libthrift-0.12.0.jar
 log4j/1.2.17//log4j-1.2.17.jar
 logging-interceptor/3.12.12//logging-interceptor-3.12.12.jar
 lz4-java/1.7.1//lz4-java-1.7.1.jar

pom.xml

@@ -187,7 +187,7 @@
     <joda.version>2.10.10</joda.version>
     <jodd.version>3.5.2</jodd.version>
     <jsr305.version>3.0.0</jsr305.version>
-    <libthrift.version>0.12.0</libthrift.version>
+    <libthrift.version>0.14.1</libthrift.version>
     <antlr4.version>4.8</antlr4.version>
     <jpam.version>1.1</jpam.version>
     <selenium.version>3.141.59</selenium.version>

sql/hive-thriftserver/src/main/java/org/apache/hive/service/auth/HiveAuthFactory.java

@@ -56,6 +56,7 @@
 import org.apache.thrift.transport.TTransport;
 import org.apache.thrift.transport.TTransportException;
 import org.apache.thrift.transport.TTransportFactory;
+import org.apache.thrift.TConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -250,8 +251,8 @@ public static UserGroupInformation loginFromSpnegoKeytabAndReturnUGI(HiveConf hi
     }
   }
 
-  public static TTransport getSocketTransport(String host, int port, int loginTimeout) {
-    return new TSocket(host, port, loginTimeout);
+  public static TTransport getSocketTransport(String host, int port, int loginTimeout) throws TTransportException {
+    return new TSocket(new TConfiguration(), host, port, loginTimeout);
   }
 
   public static TTransport getSSLSocket(String host, int port, int loginTimeout)

sql/hive-thriftserver/src/main/java/org/apache/hive/service/auth/KerberosSaslHelper.java

@@ -30,6 +30,7 @@
 import org.apache.thrift.TProcessorFactory;
 import org.apache.thrift.transport.TSaslClientTransport;
 import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
 
 public final class KerberosSaslHelper {
 
@@ -68,7 +69,7 @@ public static TTransport createSubjectAssumedTransport(String principal,
         new TSaslClientTransport("GSSAPI", null, names[0], names[1], saslProps, null,
           underlyingTransport);
       return new TSubjectAssumingTransport(saslTransport);
-    } catch (SaslException se) {
+    } catch (SaslException | TTransportException se) {
       throw new IOException("Could not instantiate SASL transport", se);
     }
   }

sql/hive-thriftserver/src/main/java/org/apache/hive/service/auth/PlainSaslHelper.java

@@ -39,7 +39,7 @@
 import org.apache.thrift.transport.TSaslServerTransport;
 import org.apache.thrift.transport.TTransport;
 import org.apache.thrift.transport.TTransportFactory;
-
+import org.apache.thrift.transport.TTransportException;
 public final class PlainSaslHelper {
 
   public static TProcessorFactory getPlainProcessorFactory(ThriftCLIService service) {
@@ -64,7 +64,7 @@ public static TTransportFactory getPlainTransportFactory(String authTypeStr)
   }
 
   public static TTransport getPlainTransport(String username, String password,
-    TTransport underlyingTransport) throws SaslException {
+    TTransport underlyingTransport) throws SaslException, TTransportException {
     return new TSaslClientTransport("PLAIN", null, null, null, new HashMap<String, String>(),
       new PlainCallbackHandler(username, password), underlyingTransport);
   }

sql/hive-thriftserver/src/main/java/org/apache/hive/service/auth/TSetIpAddressProcessor.java

@@ -45,11 +45,12 @@ public TSetIpAddressProcessor(Iface iface) {
   }
 
   @Override
-  public boolean process(final TProtocol in, final TProtocol out) throws TException {
+  public void process(final TProtocol in, final TProtocol out) throws TException {
     setIpAddress(in);
     setUserName(in);
     try {
-      return super.process(in, out);
+       super.process(in, out);
+       return;
     } finally {
       THREAD_LOCAL_USER_NAME.remove();
       THREAD_LOCAL_IP_ADDRESS.remove();

sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/thrift/ThriftBinaryCLIService.java

@@ -90,16 +90,10 @@ protected void initializeServer() {
 
       // Server args
       int maxMessageSize = hiveConf.getIntVar(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_MAX_MESSAGE_SIZE);
-      int requestTimeout = (int) hiveConf.getTimeVar(
-          HiveConf.ConfVars.HIVE_SERVER2_THRIFT_LOGIN_TIMEOUT, TimeUnit.SECONDS);
-      int beBackoffSlotLength = (int) hiveConf.getTimeVar(
-          HiveConf.ConfVars.HIVE_SERVER2_THRIFT_LOGIN_BEBACKOFF_SLOT_LENGTH, TimeUnit.MILLISECONDS);
       TThreadPoolServer.Args sargs = new TThreadPoolServer.Args(serverSocket)
           .processorFactory(processorFactory).transportFactory(transportFactory)
           .protocolFactory(new TBinaryProtocol.Factory())
           .inputProtocolFactory(new TBinaryProtocol.Factory(true, true, maxMessageSize, maxMessageSize))
-          .requestTimeout(requestTimeout).requestTimeoutUnit(TimeUnit.SECONDS)
-          .beBackoffSlotLength(beBackoffSlotLength).beBackoffSlotLengthUnit(TimeUnit.MILLISECONDS)
           .executorService(executorService);
 
       // TCP Server

sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/thrift/ThriftCLIService.java

@@ -83,6 +83,16 @@ public void setSessionHandle(SessionHandle sessionHandle) {
     public SessionHandle getSessionHandle() {
       return sessionHandle;
     }
+
+    @Override
+    public <T> T unwrap(Class<T> aClass) {
+      return null;
+    }
+
+    @Override
+    public boolean isWrapperFor(Class<?> aClass) {
+      return false;
+    }
   }
 
   public ThriftCLIService(CLIService service, String serviceName) {

sql/hive/src/main/java/org/apache/thrift/transport/TFramedTransport.java

@@ -0,0 +1,200 @@
+/*
+ * 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 org.apache.thrift.transport.TFramedTransport}.
+ * To fix class of org.apache.thrift.transport.TFramedTransport not found after upgrading libthrift.
+ *
+ * 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));
+    }
+}