drill-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From prog...@apache.org
Subject [2/2] drill git commit: DRILL-5721: Update based on merge conflict with DRILL-3449 Note: Resolved Merge Conflict and added certain new tests
Date Mon, 25 Sep 2017 05:16:51 GMT
DRILL-5721: Update based on merge conflict with DRILL-3449
            Note: Resolved Merge Conflict and added certain new tests

closes #919


Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/6cb626d7
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/6cb626d7
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/6cb626d7

Branch: refs/heads/master
Commit: 6cb626d78de75a1061e1381f680478ed8844123a
Parents: b06a7bd
Author: Sorabh Hamirwasia <shamirwasia@maprtech.com>
Authored: Tue Sep 19 21:58:39 2017 -0700
Committer: Paul Rogers <progers@maprtech.com>
Committed: Sun Sep 24 21:34:39 2017 -0700

----------------------------------------------------------------------
 .../work/fragment/FragmentStatusReporter.java   | 11 +--
 .../fragment/FragmentStatusReporterTest.java    | 83 +++++++++++++++++---
 2 files changed, 76 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/drill/blob/6cb626d7/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentStatusReporter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentStatusReporter.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentStatusReporter.java
index c0ecb07..c095edf 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentStatusReporter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentStatusReporter.java
@@ -26,7 +26,6 @@ import org.apache.drill.exec.proto.UserBitShared.FragmentState;
 import org.apache.drill.exec.proto.UserBitShared.MinorFragmentProfile;
 import org.apache.drill.exec.proto.helper.QueryIdHelper;
 import org.apache.drill.exec.rpc.control.ControlTunnel;
-import org.apache.drill.exec.server.Drillbit;
 
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -100,8 +99,7 @@ public class FragmentStatusReporter implements AutoCloseable {
       case FAILED:
         // shouldn't get here since fail() should be called.
       default:
-        throw new IllegalStateException(String.format("Received state changed event for unexpected
state of %s.",
-            newState));
+        throw new IllegalStateException(String.format("Received state changed event for unexpected
state of %s.", newState));
     }
   }
 
@@ -109,6 +107,7 @@ public class FragmentStatusReporter implements AutoCloseable {
   /**
    * Sends status to remote Foreman node using Control Tunnel or to Local Foreman bypassing
    * Control Tunnel and using WorkEventBus.
+   *
    * @param status
    */
   void sendStatus(final FragmentStatus status) {
@@ -116,8 +115,7 @@ public class FragmentStatusReporter implements AutoCloseable {
     DrillbitEndpoint foremanNode = foremanDrillbit.get();
 
     if (foremanNode == null) {
-      logger.warn("{}: State {} is not reported as {} is closed", QueryIdHelper.getQueryIdentifier(context.getHandle()),
-          status.getProfile().getState(), this);
+      logger.warn("{}: State {} is not reported as {} is closed", QueryIdHelper.getQueryIdentifier(context.getHandle()),
status.getProfile().getState(), this);
       return;
     }
 
@@ -144,8 +142,7 @@ public class FragmentStatusReporter implements AutoCloseable {
   }
 
   @Override
-  public void close()
-  {
+  public void close() {
     final DrillbitEndpoint foremanNode = foremanDrillbit.getAndSet(null);
     if (foremanNode != null) {
       logger.debug("Closing {}", this);

http://git-wip-us.apache.org/repos/asf/drill/blob/6cb626d7/exec/java-exec/src/test/java/org/apache/drill/exec/work/fragment/FragmentStatusReporterTest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/work/fragment/FragmentStatusReporterTest.java
b/exec/java-exec/src/test/java/org/apache/drill/exec/work/fragment/FragmentStatusReporterTest.java
index d4d198b..29cd6a2 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/work/fragment/FragmentStatusReporterTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/work/fragment/FragmentStatusReporterTest.java
@@ -18,6 +18,8 @@
  */
 package org.apache.drill.exec.work.fragment;
 
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.rpc.control.WorkEventBus;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -29,9 +31,11 @@ import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
 import org.apache.drill.exec.proto.UserBitShared.FragmentState;
 import org.apache.drill.exec.rpc.control.ControlTunnel;
 
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyZeroInteractions;
@@ -43,19 +47,31 @@ import static org.apache.drill.exec.proto.UserBitShared.FragmentState.RUNNING;
 
 public class FragmentStatusReporterTest {
 
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FragmentStatusReporterTest.class);
+  //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FragmentStatusReporterTest.class);
 
   private FragmentStatusReporter statusReporter;
-  private ControlTunnel tunnel;
+
+  private ControlTunnel foremanTunnel;
+
+  private FragmentContext context;
 
   @Before
   public void setUp() throws Exception {
-    FragmentContext context = mock(FragmentContext.class);
+    context = mock(FragmentContext.class);
     when(context.getStats()).thenReturn(mock(FragmentStats.class));
     when(context.getHandle()).thenReturn(FragmentHandle.getDefaultInstance());
     when(context.getAllocator()).thenReturn(mock(BufferAllocator.class));
-    tunnel = mock(ControlTunnel.class);
-    statusReporter = new FragmentStatusReporter(context, tunnel);
+
+    // Create 2 different endpoint such that foremanEndpoint is different than
+    // localEndpoint
+    DrillbitEndpoint localEndpoint = DrillbitEndpoint.newBuilder().setAddress("10.0.0.1").build();
+    DrillbitEndpoint foremanEndpoint = DrillbitEndpoint.newBuilder().setAddress("10.0.0.2").build();
+    foremanTunnel = mock(ControlTunnel.class);
+
+    when(context.getForemanEndpoint()).thenReturn(foremanEndpoint);
+    when(context.getIdentity()).thenReturn(localEndpoint);
+    when(context.getControlTunnel(foremanEndpoint)).thenReturn(foremanTunnel);
+    statusReporter = new FragmentStatusReporter(context);
   }
 
   @Test
@@ -72,35 +88,80 @@ public class FragmentStatusReporterTest {
         }
       }
     }
-    verify(tunnel, times(FragmentState.values().length - 2)) /* exclude SENDING and FAILED
*/
+    verify(foremanTunnel, times(FragmentState.values().length - 2)) /* exclude SENDING and
FAILED */
         .sendFragmentStatus(any(FragmentStatus.class));
   }
 
   @Test
   public void testFail() throws Exception {
     statusReporter.fail(null);
-    verify(tunnel).sendFragmentStatus(any(FragmentStatus.class));
+    verify(foremanTunnel).sendFragmentStatus(any(FragmentStatus.class));
   }
 
   @Test
   public void testClose() throws Exception {
     statusReporter.close();
-    verifyZeroInteractions(tunnel);
+    verifyZeroInteractions(foremanTunnel);
   }
 
   @Test
   public void testCloseClosed() throws Exception {
     statusReporter.close();
     statusReporter.close();
-    verifyZeroInteractions(tunnel);
+    verifyZeroInteractions(foremanTunnel);
   }
 
   @Test
   public void testStateChangedAfterClose() throws Exception {
     statusReporter.stateChanged(RUNNING);
-    verify(tunnel).sendFragmentStatus(any(FragmentStatus.class));
+    verify(foremanTunnel).sendFragmentStatus(any(FragmentStatus.class));
     statusReporter.close();
     statusReporter.stateChanged(CANCELLATION_REQUESTED);
-    verify(tunnel).sendFragmentStatus(any(FragmentStatus.class));
+    verify(foremanTunnel).sendFragmentStatus(any(FragmentStatus.class));
+  }
+
+
+  /**
+   * With LocalEndpoint and Foreman Endpoint being same node, test that status change
+   * message doesn't happen via Control Tunnel instead it happens locally via WorkEventBus
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testStateChangedLocalForeman() throws Exception {
+
+    DrillbitEndpoint localEndpoint = DrillbitEndpoint.newBuilder().setAddress("10.0.0.1").build();
+
+    when(context.getIdentity()).thenReturn(localEndpoint);
+    when(context.getForemanEndpoint()).thenReturn(localEndpoint);
+    when(context.getWorkEventbus()).thenReturn(mock(WorkEventBus.class));
+
+    statusReporter = new FragmentStatusReporter(context);
+    statusReporter.stateChanged(RUNNING);
+    verifyZeroInteractions(foremanTunnel);
+    verify(context.getWorkEventbus()).statusUpdate(any(FragmentStatus.class));
+  }
+
+  /**
+   * With LocalEndpoint and Foreman Endpoint being same node, test that after close of
+   * FragmentStatusReporter, status update doesn't happen either through Control Tunnel
+   * or through WorkEventBus.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testCloseLocalForeman() throws Exception {
+    DrillbitEndpoint localEndpoint = DrillbitEndpoint.newBuilder().setAddress("10.0.0.1").build();
+
+    when(context.getIdentity()).thenReturn(localEndpoint);
+    when(context.getForemanEndpoint()).thenReturn(localEndpoint);
+    when(context.getWorkEventbus()).thenReturn(mock(WorkEventBus.class));
+    statusReporter = new FragmentStatusReporter(context);
+
+    statusReporter.close();
+    assertTrue(statusReporter.foremanDrillbit.get() == null);
+    statusReporter.stateChanged(RUNNING);
+    verifyZeroInteractions(foremanTunnel);
+    verify(context.getWorkEventbus(), never()).statusUpdate(any(FragmentStatus.class));
   }
 }


Mime
View raw message