hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ss...@apache.org
Subject [10/10] hive git commit: HIVE-12934. Refactor llap module structure to allow for a usable client. (Siddharth Seth, reviewed by Sergey Shelukhin)
Date Mon, 01 Feb 2016 17:46:04 GMT
HIVE-12934. Refactor llap module structure to allow for a usable client. (Siddharth Seth, reviewed by Sergey Shelukhin)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/4185d9b8
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/4185d9b8
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/4185d9b8

Branch: refs/heads/master
Commit: 4185d9b8e2eecfef3b5a38899f6928fa82c01e99
Parents: 11f1e47
Author: Siddharth Seth <sseth@apache.org>
Authored: Mon Feb 1 09:45:09 2016 -0800
Committer: Siddharth Seth <sseth@apache.org>
Committed: Mon Feb 1 09:45:09 2016 -0800

----------------------------------------------------------------------
 llap-client/pom.xml                             |    87 +-
 .../hive/llap/security/LlapTokenIdentifier.java |    82 -
 .../hive/llap/security/LlapTokenProvider.java   |    27 -
 .../hive/llap/tez/LlapProtocolClientProxy.java  |   509 +
 .../tez/TestLlapDaemonProtocolClientProxy.java  |   144 +
 llap-common/pom.xml                             |   235 +
 .../daemon/rpc/LlapDaemonProtocolProtos.java    | 14553 +++++++++++++++++
 .../org/apache/hadoop/hive/llap/LlapNodeId.java |    86 +
 .../impl/LlapManagementProtocolClientImpl.java  |    82 +
 .../hive/llap/impl/LlapProtocolClientImpl.java  |   125 +
 .../llap/protocol/LlapManagementProtocolPB.java |    26 +
 .../llap/protocol/LlapProtocolBlockingPB.java   |    30 +
 .../protocol/LlapTaskUmbilicalProtocol.java     |    42 +
 .../hive/llap/security/LlapTokenIdentifier.java |    82 +
 .../hive/llap/security/LlapTokenProvider.java   |    27 +
 .../hive/llap/security/LlapTokenSelector.java   |    53 +
 .../apache/hadoop/hive/llap/tez/Converters.java |   265 +
 .../src/protobuf/LlapDaemonProtocol.proto       |   148 +
 .../hive/llap/testhelpers/ControlledClock.java  |    43 +
 llap-server/pom.xml                             |   199 +-
 .../daemon/rpc/LlapDaemonProtocolProtos.java    | 14553 -----------------
 .../org/apache/hadoop/hive/llap/LlapNodeId.java |    86 -
 .../hadoop/hive/llap/cli/LlapServiceDriver.java |     8 +
 .../daemon/LlapDaemonProtocolBlockingPB.java    |    28 -
 .../LlapManagementProtocolBlockingPB.java       |    24 -
 .../hive/llap/daemon/impl/LlapDaemon.java       |     6 +-
 .../impl/LlapDaemonProtocolClientImpl.java      |   125 -
 .../impl/LlapDaemonProtocolServerImpl.java      |   251 -
 .../impl/LlapManagementProtocolClientImpl.java  |    82 -
 .../daemon/impl/LlapProtocolServerImpl.java     |   251 +
 .../llap/daemon/impl/QueryFragmentInfo.java     |    13 +-
 .../llap/daemon/impl/TaskRunnerCallable.java    |     2 +-
 .../protocol/LlapTaskUmbilicalProtocol.java     |    42 -
 .../llap/security/LlapDaemonPolicyProvider.java |     8 +-
 .../hive/llap/security/LlapSecurityHelper.java  |     2 +-
 .../llap/security/LlapServerSecurityInfo.java   |    10 +-
 .../hive/llap/security/LlapTokenSelector.java   |    53 -
 .../hadoop/hive/llap/tezplugins/Converters.java |   265 -
 .../llap/tezplugins/LlapContainerLauncher.java  |    43 -
 .../LlapDaemonProtocolClientProxy.java          |   509 -
 .../llap/tezplugins/LlapTaskCommunicator.java   |   755 -
 .../tezplugins/LlapUmbilicalPolicyProvider.java |    42 -
 .../tezplugins/helpers/SourceStateTracker.java  |   296 -
 .../apache/tez/dag/app/rm/ContainerFactory.java |    51 -
 .../dag/app/rm/LlapTaskSchedulerService.java    |  1512 --
 .../src/protobuf/LlapDaemonProtocol.proto       |   148 -
 .../impl/TestLlapDaemonProtocolServerImpl.java  |    11 +-
 .../TestLlapDaemonProtocolClientProxy.java      |   143 -
 .../tezplugins/TestLlapTaskCommunicator.java    |   100 -
 .../app/rm/TestLlapTaskSchedulerService.java    |   685 -
 llap-tez/pom.xml                                |   200 +
 .../hive/llap/tezplugins/ContainerFactory.java  |    51 +
 .../llap/tezplugins/LlapContainerLauncher.java  |    43 +
 .../llap/tezplugins/LlapTaskCommunicator.java   |   757 +
 .../tezplugins/LlapTaskSchedulerService.java    |  1512 ++
 .../hive/llap/tezplugins/LlapTezUtils.java      |    29 +
 .../tezplugins/LlapUmbilicalPolicyProvider.java |    42 +
 .../tezplugins/helpers/SourceStateTracker.java  |   289 +
 .../tezplugins/TestLlapTaskCommunicator.java    |   100 +
 .../TestLlapTaskSchedulerService.java           |   684 +
 pom.xml                                         |     4 +-
 ql/pom.xml                                      |    54 +
 .../hive/ql/exec/tez/TezSessionState.java       |    58 +-
 63 files changed, 20631 insertions(+), 20141 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-client/pom.xml
----------------------------------------------------------------------
diff --git a/llap-client/pom.xml b/llap-client/pom.xml
index f6a5629..50c06a4 100644
--- a/llap-client/pom.xml
+++ b/llap-client/pom.xml
@@ -32,7 +32,7 @@
   </properties>
 
   <dependencies>
-    <!-- dependencies are always listed in sorted order by groupId, artifectId -->
+    <!-- dependencies are always listed in sorted order by groupId, artifactId -->
     <!-- intra-project -->
     <dependency>
       <groupId>org.apache.hive</groupId>
@@ -41,41 +41,26 @@
     </dependency>
     <dependency>
       <groupId>org.apache.hive</groupId>
-      <artifactId>hive-serde</artifactId>
+      <artifactId>hive-llap-common</artifactId>
       <version>${project.version}</version>
     </dependency>
     <!-- inter-project -->
     <dependency>
-      <groupId>commons-codec</groupId>
-      <artifactId>commons-codec</artifactId>
-      <version>${commons-codec.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>commons-lang</groupId>
-      <artifactId>commons-lang</artifactId>
-      <version>${commons-lang.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.thrift</groupId>
-      <artifactId>libthrift</artifactId>
-      <version>${libthrift.version}</version>
-    </dependency>
-    <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>
       <version>${hadoop.version}</version>
       <optional>true</optional>
-          <exclusions>
-            <exclusion>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-          </exclusion>
-          <exclusion>
-            <groupId>commmons-logging</groupId>
-            <artifactId>commons-logging</artifactId>
-          </exclusion>
-        </exclusions>
-   </dependency>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commmons-logging</groupId>
+          <artifactId>commons-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-mapreduce-client-core</artifactId>
@@ -96,15 +81,15 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.mockito</groupId>
-      <artifactId>mockito-all</artifactId>
-      <version>${mockito-all.version}</version>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>${commons-lang3.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>com.sun.jersey</groupId>
-      <artifactId>jersey-servlet</artifactId>
-      <version>${jersey.version}</version>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+      <version>${mockito-all.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -113,32 +98,18 @@
       <version>${hadoop.version}</version>
       <classifier>tests</classifier>
       <scope>test</scope>
-          <exclusions>
-            <exclusion>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-          </exclusion>
-          <exclusion>
-            <groupId>commmons-logging</groupId>
-            <artifactId>commons-logging</artifactId>
-          </exclusion>
-        </exclusions>
-   </dependency>
-    <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-hdfs</artifactId>
-      <version>${hadoop.version}</version>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-hdfs</artifactId>
-      <version>${hadoop.version}</version>
-      <classifier>tests</classifier>
-      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commmons-logging</groupId>
+          <artifactId>commons-logging</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
   </dependencies>
-
   <build>
     <sourceDirectory>${basedir}/src/java</sourceDirectory>
     <testSourceDirectory>${basedir}/src/test</testSourceDirectory>

http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java b/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java
deleted file mode 100644
index f0bb495..0000000
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenIdentifier.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/**
- * 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.hadoop.hive.llap.security;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdentifier;
-
-/** For now, a LLAP token gives access to any LLAP server. */
-public class LlapTokenIdentifier extends AbstractDelegationTokenIdentifier {
-  private static final String KIND = "LLAP_TOKEN";
-  public static final Text KIND_NAME = new Text(KIND);
-
-  public LlapTokenIdentifier() {
-    super();
-  }
-
-  public LlapTokenIdentifier(Text owner, Text renewer, Text realUser) {
-    super(owner, renewer, realUser);
-  }
-
-  @Override
-  public void write(DataOutput out) throws IOException {
-    super.write(out);
-    // Nothing right now.
-  }
-
-  @Override
-  public void readFields(DataInput in) throws IOException {
-    super.readFields(in);
-    // Nothing right now.
-  }
-
-  @Override
-  public Text getKind() {
-    return KIND_NAME;
-  }
-
-  @Override
-  public int hashCode() {
-    return -1;
-  }
-
-  @Override
-  public boolean equals(Object other) {
-    return (other != null) && other.getClass().isAssignableFrom(this.getClass());
-  }
-
-  @Override
-  public String toString() {
-    return KIND;
-  }
-
-  @InterfaceAudience.Private
-  public static class Renewer extends Token.TrivialRenewer {
-    @Override
-    protected Text getKind() {
-      return KIND_NAME;
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java b/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java
deleted file mode 100644
index 2e99a28..0000000
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/security/LlapTokenProvider.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/**
- * 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.hadoop.hive.llap.security;
-
-import java.io.IOException;
-
-import org.apache.hadoop.security.token.Token;
-
-public interface LlapTokenProvider {
-  Token<LlapTokenIdentifier> getDelegationToken() throws IOException;
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java b/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java
new file mode 100644
index 0000000..5b0674a
--- /dev/null
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/tez/LlapProtocolClientProxy.java
@@ -0,0 +1,509 @@
+/*
+ * Licensed 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.hadoop.hive.llap.tez;
+
+import javax.net.SocketFactory;
+
+import java.io.IOException;
+import java.security.PrivilegedAction;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.protobuf.Message;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.llap.LlapNodeId;
+import org.apache.hadoop.hive.llap.protocol.LlapProtocolBlockingPB;
+import org.apache.hadoop.hive.llap.impl.LlapProtocolClientImpl;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.QueryCompleteResponseProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SourceStateUpdatedResponseProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkResponseProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentRequestProto;
+import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.TerminateFragmentResponseProto;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.service.AbstractService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LlapProtocolClientProxy extends AbstractService {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LlapProtocolClientProxy.class);
+
+  private final ConcurrentMap<String, LlapProtocolBlockingPB> hostProxies;
+
+  private final RequestManager requestManager;
+  private final RetryPolicy retryPolicy;
+  private final SocketFactory socketFactory;
+
+  private final ListeningExecutorService requestManagerExecutor;
+  private volatile ListenableFuture<Void> requestManagerFuture;
+  private final Token<LlapTokenIdentifier> llapToken;
+
+  public LlapProtocolClientProxy(
+      int numThreads, Configuration conf, Token<LlapTokenIdentifier> llapToken) {
+    super(LlapProtocolClientProxy.class.getSimpleName());
+    this.hostProxies = new ConcurrentHashMap<>();
+    this.socketFactory = NetUtils.getDefaultSocketFactory(conf);
+    this.llapToken = llapToken;
+
+    long connectionTimeout = HiveConf.getTimeVar(conf,
+        ConfVars.LLAP_TASK_COMMUNICATOR_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    long retrySleep = HiveConf.getTimeVar(conf,
+        ConfVars.LLAP_TASK_COMMUNICATOR_CONNECTION_SLEEP_BETWEEN_RETRIES_MS,
+        TimeUnit.MILLISECONDS);
+    this.retryPolicy = RetryPolicies.retryUpToMaximumTimeWithFixedSleep(
+        connectionTimeout, retrySleep, TimeUnit.MILLISECONDS);
+
+    this.requestManager = new RequestManager(numThreads);
+    ExecutorService localExecutor = Executors.newFixedThreadPool(1,
+        new ThreadFactoryBuilder().setNameFormat("RequestManagerExecutor").build());
+    this.requestManagerExecutor = MoreExecutors.listeningDecorator(localExecutor);
+
+    LOG.info("Setting up taskCommunicator with" +
+        "numThreads=" + numThreads +
+        "retryTime(millis)=" + connectionTimeout +
+        "retrySleep(millis)=" + retrySleep);
+  }
+
+  @Override
+  public void serviceStart() {
+    requestManagerFuture = requestManagerExecutor.submit(requestManager);
+    Futures.addCallback(requestManagerFuture, new FutureCallback<Void>() {
+      @Override
+      public void onSuccess(Void result) {
+        LOG.info("RequestManager shutdown");
+      }
+
+      @Override
+      public void onFailure(Throwable t) {
+        LOG.warn("RequestManager shutdown with error", t);
+      }
+    });
+  }
+
+  @Override
+  public void serviceStop() {
+    if (requestManagerFuture != null) {
+      requestManager.shutdown();
+      requestManagerFuture.cancel(true);
+    }
+    requestManagerExecutor.shutdown();
+  }
+
+  public void sendSubmitWork(SubmitWorkRequestProto request, String host, int port,
+                         final ExecuteRequestCallback<SubmitWorkResponseProto> callback) {
+    LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
+    requestManager.queueRequest(new SubmitWorkCallable(nodeId, request, callback));
+  }
+
+  public void sendSourceStateUpdate(final SourceStateUpdatedRequestProto request, final String host,
+                                    final int port,
+                                    final ExecuteRequestCallback<SourceStateUpdatedResponseProto> callback) {
+    LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
+    requestManager.queueRequest(
+        new SendSourceStateUpdateCallable(nodeId, request, callback));
+  }
+
+  public void sendQueryComplete(final QueryCompleteRequestProto request, final String host,
+                                final int port,
+                                final ExecuteRequestCallback<QueryCompleteResponseProto> callback) {
+    LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
+    requestManager.queueRequest(new SendQueryCompleteCallable(nodeId, request, callback));
+  }
+
+  public void sendTerminateFragment(final TerminateFragmentRequestProto request, final String host,
+                                    final int port,
+                                    final ExecuteRequestCallback<TerminateFragmentResponseProto> callback) {
+    LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
+    requestManager.queueRequest(new SendTerminateFragmentCallable(nodeId, request, callback));
+  }
+
+  @VisibleForTesting
+  static class RequestManager implements Callable<Void> {
+
+    private final Lock lock = new ReentrantLock();
+    private final AtomicBoolean isShutdown = new AtomicBoolean(false);
+    private final Condition queueCondition = lock.newCondition();
+    private final AtomicBoolean shouldRun = new AtomicBoolean(false);
+
+    private final int maxConcurrentRequestsPerNode = 1;
+    private final ListeningExecutorService executor;
+
+
+    // Tracks new additions via add, while the loop is processing existing ones.
+    private final LinkedList<CallableRequest> newRequestList = new LinkedList<>();
+
+    // Tracks existing requests which are cycled through.
+    private final LinkedList<CallableRequest> pendingRequests = new LinkedList<>();
+
+    // Tracks requests executing per node
+    private final ConcurrentMap<LlapNodeId, AtomicInteger> runningRequests = new ConcurrentHashMap<>();
+
+    // Tracks completed requests pre node
+    private final LinkedList<LlapNodeId> completedNodes = new LinkedList<>();
+
+    public RequestManager(int numThreads) {
+      ExecutorService localExecutor = Executors.newFixedThreadPool(numThreads,
+          new ThreadFactoryBuilder().setNameFormat("TaskCommunicator #%2d").build());
+      executor = MoreExecutors.listeningDecorator(localExecutor);
+    }
+
+
+    @VisibleForTesting
+    Set<LlapNodeId> currentLoopDisabledNodes = new HashSet<>();
+    @VisibleForTesting
+    List<CallableRequest> currentLoopSkippedRequests = new LinkedList<>();
+    @Override
+    public Void call() {
+      // Caches disabled nodes for quicker lookups and ensures a request on a node which was skipped
+      // does not go out of order.
+      while (!isShutdown.get()) {
+        lock.lock();
+        try {
+          while (!shouldRun.get()) {
+            queueCondition.await();
+            break; // Break out and try executing.
+          }
+          boolean shouldBreak = process();
+          if (shouldBreak) {
+            break;
+          }
+        } catch (InterruptedException e) {
+          if (isShutdown.get()) {
+            break;
+          } else {
+            LOG.warn("RunLoop interrupted without being shutdown first");
+            throw new RuntimeException(e);
+          }
+        } finally {
+          lock.unlock();
+        }
+      }
+      LOG.info("CallScheduler loop exiting");
+      return null;
+    }
+
+    /* Add a new request to be executed */
+    public void queueRequest(CallableRequest request) {
+      synchronized (newRequestList) {
+        newRequestList.add(request);
+        shouldRun.set(true);
+      }
+      notifyRunLoop();
+    }
+
+    /* Indicates a request has completed on a node */
+    public void requestFinished(LlapNodeId nodeId) {
+      synchronized (completedNodes) {
+        completedNodes.add(nodeId);
+        shouldRun.set(true);
+      }
+      notifyRunLoop();
+    }
+
+    public void shutdown() {
+      if (!isShutdown.getAndSet(true)) {
+        executor.shutdownNow();
+        notifyRunLoop();
+      }
+    }
+
+    @VisibleForTesting
+    void submitToExecutor(CallableRequest request, LlapNodeId nodeId) {
+      ListenableFuture<SourceStateUpdatedResponseProto> future =
+          executor.submit(request);
+      Futures.addCallback(future, new ResponseCallback(request.getCallback(), nodeId, this));
+    }
+
+    @VisibleForTesting
+    boolean process() {
+      if (isShutdown.get()) {
+        return true;
+      }
+      currentLoopDisabledNodes.clear();
+      currentLoopSkippedRequests.clear();
+
+      // Set to false to block the next loop. This must be called before draining the lists,
+      // otherwise an add/completion after draining the lists but before setting it to false,
+      // will not trigger a run. May cause one unnecessary run if an add comes in before drain.
+      // drain list. add request (setTrue). setFalse needs to be avoided.
+      shouldRun.compareAndSet(true, false);
+      // Drain any calls which may have come in during the last execution of the loop.
+      drainNewRequestList();  // Locks newRequestList
+      drainCompletedNodes();  // Locks completedNodes
+
+
+      Iterator<CallableRequest> iterator = pendingRequests.iterator();
+      while (iterator.hasNext()) {
+        CallableRequest request = iterator.next();
+        iterator.remove();
+        LlapNodeId nodeId = request.getNodeId();
+        if (canRunForNode(nodeId, currentLoopDisabledNodes)) {
+          submitToExecutor(request, nodeId);
+        } else {
+          currentLoopDisabledNodes.add(nodeId);
+          currentLoopSkippedRequests.add(request);
+        }
+      }
+      // Tried scheduling everything that could be scheduled in this loop.
+      pendingRequests.addAll(0, currentLoopSkippedRequests);
+      return false;
+    }
+
+    private void drainNewRequestList() {
+      synchronized (newRequestList) {
+        if (!newRequestList.isEmpty()) {
+          pendingRequests.addAll(newRequestList);
+          newRequestList.clear();
+        }
+      }
+    }
+
+    private void drainCompletedNodes() {
+      synchronized (completedNodes) {
+        if (!completedNodes.isEmpty()) {
+          for (LlapNodeId nodeId : completedNodes) {
+            runningRequests.get(nodeId).decrementAndGet();
+          }
+        }
+        completedNodes.clear();
+      }
+    }
+
+    private boolean canRunForNode(LlapNodeId nodeId, Set<LlapNodeId> currentRunDisabledNodes) {
+      if (currentRunDisabledNodes.contains(nodeId)) {
+        return false;
+      } else {
+        AtomicInteger count = runningRequests.get(nodeId);
+        if (count == null) {
+          count = new AtomicInteger(0);
+          AtomicInteger old = runningRequests.putIfAbsent(nodeId, count);
+          count = old != null ? old : count;
+        }
+        if (count.incrementAndGet() <= maxConcurrentRequestsPerNode) {
+          return true;
+        } else {
+          count.decrementAndGet();
+          return false;
+        }
+      }
+    }
+
+    private void notifyRunLoop() {
+      lock.lock();
+      try {
+        queueCondition.signal();
+      } finally {
+        lock.unlock();
+      }
+    }
+  }
+
+
+  private static final class ResponseCallback<TYPE extends Message>
+      implements FutureCallback<TYPE> {
+
+    private final ExecuteRequestCallback<TYPE> callback;
+    private final LlapNodeId nodeId;
+    private final RequestManager requestManager;
+
+    public ResponseCallback(ExecuteRequestCallback<TYPE> callback, LlapNodeId nodeId,
+                            RequestManager requestManager) {
+      this.callback = callback;
+      this.nodeId = nodeId;
+      this.requestManager = requestManager;
+    }
+
+    @Override
+    public void onSuccess(TYPE result) {
+      try {
+        callback.setResponse(result);
+      } finally {
+        requestManager.requestFinished(nodeId);
+      }
+    }
+
+    @Override
+    public void onFailure(Throwable t) {
+      try {
+        callback.indicateError(t);
+      } finally {
+        requestManager.requestFinished(nodeId);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  static abstract class CallableRequest<REQUEST extends Message, RESPONSE extends Message>
+      implements Callable {
+
+    final LlapNodeId nodeId;
+    final ExecuteRequestCallback<RESPONSE> callback;
+    final REQUEST request;
+
+
+    protected CallableRequest(LlapNodeId nodeId, REQUEST request, ExecuteRequestCallback<RESPONSE> callback) {
+      this.nodeId = nodeId;
+      this.request = request;
+      this.callback = callback;
+    }
+
+    public LlapNodeId getNodeId() {
+      return nodeId;
+    }
+
+    public ExecuteRequestCallback<RESPONSE> getCallback() {
+      return callback;
+    }
+
+    public abstract RESPONSE call() throws Exception;
+  }
+
+  private class SubmitWorkCallable extends CallableRequest<SubmitWorkRequestProto, SubmitWorkResponseProto> {
+
+    protected SubmitWorkCallable(LlapNodeId nodeId,
+                          SubmitWorkRequestProto submitWorkRequestProto,
+                                 ExecuteRequestCallback<SubmitWorkResponseProto> callback) {
+      super(nodeId, submitWorkRequestProto, callback);
+    }
+
+    @Override
+    public SubmitWorkResponseProto call() throws Exception {
+      return getProxy(nodeId).submitWork(null, request);
+    }
+  }
+
+  private class SendSourceStateUpdateCallable
+      extends CallableRequest<SourceStateUpdatedRequestProto, SourceStateUpdatedResponseProto> {
+
+    public SendSourceStateUpdateCallable(LlapNodeId nodeId,
+                                         SourceStateUpdatedRequestProto request,
+                                         ExecuteRequestCallback<SourceStateUpdatedResponseProto> callback) {
+      super(nodeId, request, callback);
+    }
+
+    @Override
+    public SourceStateUpdatedResponseProto call() throws Exception {
+      return getProxy(nodeId).sourceStateUpdated(null, request);
+    }
+  }
+
+  private class SendQueryCompleteCallable
+      extends CallableRequest<QueryCompleteRequestProto, QueryCompleteResponseProto> {
+
+    protected SendQueryCompleteCallable(LlapNodeId nodeId,
+                                        QueryCompleteRequestProto queryCompleteRequestProto,
+                                        ExecuteRequestCallback<QueryCompleteResponseProto> callback) {
+      super(nodeId, queryCompleteRequestProto, callback);
+    }
+
+    @Override
+    public QueryCompleteResponseProto call() throws Exception {
+      return getProxy(nodeId).queryComplete(null, request);
+    }
+  }
+
+  private class SendTerminateFragmentCallable
+      extends CallableRequest<TerminateFragmentRequestProto, TerminateFragmentResponseProto> {
+
+    protected SendTerminateFragmentCallable(LlapNodeId nodeId,
+                                            TerminateFragmentRequestProto terminateFragmentRequestProto,
+                                            ExecuteRequestCallback<TerminateFragmentResponseProto> callback) {
+      super(nodeId, terminateFragmentRequestProto, callback);
+    }
+
+    @Override
+    public TerminateFragmentResponseProto call() throws Exception {
+      return getProxy(nodeId).terminateFragment(null, request);
+    }
+  }
+
+  public interface ExecuteRequestCallback<T extends Message> {
+    void setResponse(T response);
+    void indicateError(Throwable t);
+  }
+
+  private LlapProtocolBlockingPB getProxy(final LlapNodeId nodeId) {
+    String hostId = getHostIdentifier(nodeId.getHostname(), nodeId.getPort());
+
+    LlapProtocolBlockingPB proxy = hostProxies.get(hostId);
+    if (proxy == null) {
+      if (llapToken == null) {
+        proxy = new LlapProtocolClientImpl(getConfig(), nodeId.getHostname(),
+            nodeId.getPort(), retryPolicy, socketFactory);
+      } else {
+        UserGroupInformation ugi;
+        try {
+          ugi = UserGroupInformation.getCurrentUser();
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+        Token<LlapTokenIdentifier> nodeToken = new Token<LlapTokenIdentifier>(llapToken);
+        SecurityUtil.setTokenService(nodeToken, NetUtils.createSocketAddrForHost(
+            nodeId.getHostname(), nodeId.getPort()));
+        ugi.addToken(nodeToken);
+        proxy = ugi.doAs(new PrivilegedAction<LlapProtocolBlockingPB>() {
+          @Override
+          public LlapProtocolBlockingPB run() {
+           return new LlapProtocolClientImpl(getConfig(), nodeId.getHostname(),
+               nodeId.getPort(), retryPolicy, socketFactory);
+          }
+        });
+      }
+
+      LlapProtocolBlockingPB proxyOld = hostProxies.putIfAbsent(hostId, proxy);
+      if (proxyOld != null) {
+        // TODO Shutdown the new proxy.
+        proxy = proxyOld;
+      }
+    }
+    return proxy;
+  }
+
+  private String getHostIdentifier(String hostname, int port) {
+    return hostname + ":" + port;
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-client/src/test/org/apache/hadoop/hive/llap/tez/TestLlapDaemonProtocolClientProxy.java
----------------------------------------------------------------------
diff --git a/llap-client/src/test/org/apache/hadoop/hive/llap/tez/TestLlapDaemonProtocolClientProxy.java b/llap-client/src/test/org/apache/hadoop/hive/llap/tez/TestLlapDaemonProtocolClientProxy.java
new file mode 100644
index 0000000..850b67a
--- /dev/null
+++ b/llap-client/src/test/org/apache/hadoop/hive/llap/tez/TestLlapDaemonProtocolClientProxy.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed 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.hadoop.hive.llap.tez;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.protobuf.Message;
+import org.apache.commons.lang3.mutable.MutableInt;
+import org.apache.hadoop.hive.llap.LlapNodeId;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestLlapDaemonProtocolClientProxy {
+
+  @Test (timeout = 5000)
+  public void testMultipleNodes() {
+    RequestManagerForTest requestManager = new RequestManagerForTest(1);
+
+    LlapNodeId nodeId1 = LlapNodeId.getInstance("host1", 1025);
+    LlapNodeId nodeId2 = LlapNodeId.getInstance("host2", 1025);
+
+    Message mockMessage = mock(Message.class);
+    LlapProtocolClientProxy.ExecuteRequestCallback mockExecuteRequestCallback = mock(
+        LlapProtocolClientProxy.ExecuteRequestCallback.class);
+
+    // Request two messages
+    requestManager.queueRequest(
+        new CallableRequestForTest(nodeId1, mockMessage, mockExecuteRequestCallback));
+    requestManager.queueRequest(
+        new CallableRequestForTest(nodeId2, mockMessage, mockExecuteRequestCallback));
+
+    // Should go through in a single process call
+    requestManager.process();
+    assertEquals(2, requestManager.numSubmissionsCounters);
+    assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
+    assertNotNull(requestManager.numInvocationsPerNode.get(nodeId2));
+    Assert.assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
+    Assert.assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId2).getValue().intValue());
+    assertEquals(0, requestManager.currentLoopSkippedRequests.size());
+    assertEquals(0, requestManager.currentLoopSkippedRequests.size());
+    assertEquals(0, requestManager.currentLoopDisabledNodes.size());
+  }
+
+  @Test(timeout = 5000)
+  public void testSingleInvocationPerNode() {
+    RequestManagerForTest requestManager = new RequestManagerForTest(1);
+
+    LlapNodeId nodeId1 = LlapNodeId.getInstance("host1", 1025);
+
+    Message mockMessage = mock(Message.class);
+    LlapProtocolClientProxy.ExecuteRequestCallback mockExecuteRequestCallback = mock(
+        LlapProtocolClientProxy.ExecuteRequestCallback.class);
+
+    // First request for host.
+    requestManager.queueRequest(
+        new CallableRequestForTest(nodeId1, mockMessage, mockExecuteRequestCallback));
+    requestManager.process();
+    assertEquals(1, requestManager.numSubmissionsCounters);
+    assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
+    Assert.assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
+    assertEquals(0, requestManager.currentLoopSkippedRequests.size());
+
+    // Second request for host. Single invocation since the last has not completed.
+    requestManager.queueRequest(
+        new CallableRequestForTest(nodeId1, mockMessage, mockExecuteRequestCallback));
+    requestManager.process();
+    assertEquals(1, requestManager.numSubmissionsCounters);
+    assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
+    Assert.assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
+    assertEquals(1, requestManager.currentLoopSkippedRequests.size());
+    assertEquals(1, requestManager.currentLoopDisabledNodes.size());
+    assertTrue(requestManager.currentLoopDisabledNodes.contains(nodeId1));
+
+    // Complete first request. Second pending request should go through.
+    requestManager.requestFinished(nodeId1);
+    requestManager.process();
+    assertEquals(2, requestManager.numSubmissionsCounters);
+    assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
+    Assert.assertEquals(2, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
+    assertEquals(0, requestManager.currentLoopSkippedRequests.size());
+    assertEquals(0, requestManager.currentLoopDisabledNodes.size());
+    assertFalse(requestManager.currentLoopDisabledNodes.contains(nodeId1));
+  }
+
+
+  static class RequestManagerForTest extends LlapProtocolClientProxy.RequestManager {
+
+    int numSubmissionsCounters = 0;
+    private Map<LlapNodeId, MutableInt> numInvocationsPerNode = new HashMap<>();
+
+    public RequestManagerForTest(int numThreads) {
+      super(numThreads);
+    }
+
+    protected void submitToExecutor(LlapProtocolClientProxy.CallableRequest request, LlapNodeId nodeId) {
+      numSubmissionsCounters++;
+      MutableInt nodeCount = numInvocationsPerNode.get(nodeId);
+      if (nodeCount == null) {
+        nodeCount = new MutableInt(0);
+        numInvocationsPerNode.put(nodeId, nodeCount);
+      }
+      nodeCount.increment();
+    }
+
+    void reset() {
+      numSubmissionsCounters = 0;
+      numInvocationsPerNode.clear();
+    }
+
+  }
+
+  static class CallableRequestForTest extends LlapProtocolClientProxy.CallableRequest<Message, Message> {
+
+    protected CallableRequestForTest(LlapNodeId nodeId, Message message,
+                                     LlapProtocolClientProxy.ExecuteRequestCallback<Message> callback) {
+      super(nodeId, message, callback);
+    }
+
+    @Override
+    public Message call() throws Exception {
+      return null;
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-common/pom.xml
----------------------------------------------------------------------
diff --git a/llap-common/pom.xml b/llap-common/pom.xml
new file mode 100644
index 0000000..5343479
--- /dev/null
+++ b/llap-common/pom.xml
@@ -0,0 +1,235 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.apache.hive</groupId>
+    <artifactId>hive</artifactId>
+    <version>2.1.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>hive-llap-common</artifactId>
+  <packaging>jar</packaging>
+  <name>Hive Llap Common</name>
+
+  <properties>
+    <hive.path.to.root>..</hive.path.to.root>
+  </properties>
+
+  <dependencies>
+    <!-- dependencies are always listed in sorted order by groupId, artifactId -->
+    <!-- intra-project -->
+    <dependency>
+      <groupId>org.apache.hive</groupId>
+      <artifactId>hive-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <!-- inter-project -->
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
+      <version>${hadoop.version}</version>
+      <optional>true</optional>
+            <exclusions>
+            <exclusion>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>commmons-logging</groupId>
+            <artifactId>commons-logging</artifactId>
+          </exclusion>
+        </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.tez</groupId>
+      <artifactId>tez-api</artifactId>
+      <version>${tez.version}</version>
+      <optional>true</optional>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commmons-logging</groupId>
+          <artifactId>commons-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.tez</groupId>
+      <artifactId>tez-common</artifactId>
+      <version>${tez.version}</version>
+      <optional>true</optional>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commmons-logging</groupId>
+          <artifactId>commons-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.tez</groupId>
+      <artifactId>tez-runtime-internals</artifactId>
+      <version>${tez.version}</version>
+      <optional>true</optional>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commmons-logging</groupId>
+          <artifactId>commons-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <!-- test inter-project -->
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
+      <version>${hadoop.version}</version>
+      <classifier>tests</classifier>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commmons-logging</groupId>
+          <artifactId>commons-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>${junit.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+      <version>${mockito-all.version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <profiles>
+    <profile>
+      <id>protobuf</id>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-antrun-plugin</artifactId>
+            <executions>
+              <execution>
+                <id>generate-protobuf-sources</id>
+                <phase>generate-sources</phase>
+                <configuration>
+                  <target>
+                    <property name="protobuf.src.dir"  location="${basedir}/src/protobuf"/>
+                    <property name="protobuf.build.dir"  location="${basedir}/src/gen/protobuf/gen-java"/>
+                    <echo>Building LLAP Protobuf</echo>
+                    <mkdir dir="${protobuf.build.dir}"/>
+                    <exec executable="protoc" failonerror="true">
+                      <arg value="--java_out=${protobuf.build.dir}"/>
+                      <arg value="-I=${protobuf.src.dir}"/>
+                      <arg value="${protobuf.src.dir}/LlapDaemonProtocol.proto"/>
+                    </exec>
+                  </target>
+                </configuration>
+                <goals>
+                  <goal>run</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-jar-plugin</artifactId>
+            <executions>
+              <execution>
+                <goals>
+                  <goal>test-jar</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+
+
+  <build>
+    <sourceDirectory>${basedir}/src/java</sourceDirectory>
+    <testSourceDirectory>${basedir}/src/test</testSourceDirectory>
+    <resources>
+    <resource>
+      <directory>src/main/resources</directory>
+      <excludes>
+        <exclude>*.py</exclude>
+        <exclude>*.pyc</exclude>
+      </excludes>
+      <filtering>false</filtering>
+    </resource>
+    </resources>
+    <plugins>
+      <plugin>
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>build-helper-maven-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>add-source</id>
+            <phase>generate-sources</phase>
+            <goals>
+              <goal>add-source</goal>
+            </goals>
+            <configuration>
+              <sources>
+                <source>src/gen/protobuf/gen-java</source>
+                <source>src/gen/thrift/gen-javabean</source>
+              </sources>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-jar-plugin</artifactId>
+        <executions>
+          <execution>
+            <goals>
+              <goal>test-jar</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>


Mime
View raw message