DRILL-4586: Add ErrorType#CLIENT to UserException for client side errors

common/src/main/java/org/apache/drill/common/exceptions/UserException.java

@@ -351,6 +351,32 @@ public static Builder unsupportedError(final Throwable cause) {
     return new Builder(DrillPBError.ErrorType.UNSUPPORTED_OPERATION, cause);
   }
 
+  /**
+   * Creates a new user exception builder.
+   *
+   * @see org.apache.drill.exec.proto.UserBitShared.DrillPBError.ErrorType#CLIENT
+   * @return user exception builder
+   */
+  public static Builder clientError() {
+    return clientError(null);
+  }
+
+  /**
+   * Wraps the passed exception inside a client error.
+   * <p>The cause message will be used unless {@link Builder#message(String, Object...)} is called.
+   * <p>If the wrapped exception is, or wraps, a user exception it will be returned by {@link Builder#build(Logger)}
+   * instead of creating a new exception. Any added context will be added to the user exception as well.
+   *
+   * @see org.apache.drill.exec.proto.UserBitShared.DrillPBError.ErrorType#CLIENT
+   *
+   * @param cause exception we want the user exception to wrap. If cause is, or wrap, a user exception it will be
+   *              returned by the builder instead of creating a new user exception
+   * @return user exception builder
+   */
+  public static Builder clientError(final Throwable cause) {
+    return new Builder(DrillPBError.ErrorType.CLIENT, cause);
+  }
+
   /**
    * Builder class for DrillUserException. You can wrap an existing exception, in this case it will first check if
    * this exception is, or wraps, a DrillUserException. If it does then the builder will use the user exception as it is

exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java

@@ -128,7 +128,7 @@ public void resultArrived( ByteBuf pBody ) throws RpcException {
         try {
           resultsListener.queryCompleted(queryState);
         } catch ( Exception e ) {
-          resultsListener.submissionFailed(UserException.systemError(e).build(logger));
+          resultsListener.submissionFailed(UserException.clientError(e).build(logger));
         }
       } else {
         logger.warn("queryState {} was ignored", queryState);
@@ -171,7 +171,7 @@ public void batchArrived( ConnectionThrottle throttle,
       // That releases batch if successful.
     } catch ( Exception e ) {
       batch.release();
-      resultsListener.submissionFailed(UserException.systemError(e).build(logger));
+      resultsListener.submissionFailed(UserException.clientError(e).build(logger));
     }
   }
 
@@ -290,8 +290,7 @@ public void failed(RpcException ex) {
       // However, the results could not be transferred to this resultListener because
       // there is no query id mapped to this resultListener. Look out for the warning
       // message from ChannelClosedHandler in the client logs.
-      // TODO(DRILL-4586)
-      resultsListener.submissionFailed(UserException.systemError(ex)
+      resultsListener.submissionFailed(UserException.clientError(ex)
           .addContext("Query submission to Drillbit failed.")
           .build(logger));
     }
@@ -341,8 +340,7 @@ public void interrupted(final InterruptedException ex) {
         return;
       }
 
-      // TODO(DRILL-4586)
-      resultsListener.submissionFailed(UserException.systemError(ex)
+      resultsListener.submissionFailed(UserException.clientError(ex)
           .addContext("The client had been asked to wait as the Drillbit is potentially being over-utilized." +
               " But the client was interrupted while waiting.")
           .build(logger));