IMPALA-12785: Add commands to control event-processor status

This patch extends the existing AdminFnStmt to support operations on
EventProcessor. E.g. to pause the EventProcessor:
  impala-shell> :event_processor('pause');
to resume the EventProcessor:
  impala-shell> :event_processor('start');
Or to resume the EventProcessor on a given event id (1000):
  impala-shell> :event_processor('start', 1000);
Admin can also resume the EventProcessor at the latest event id by using
-1:
  impala-shell> :event_processor('start', -1);

Supported command actions in this patch: pause, start, status.

The command output of all actions will show the latest status of
EventProcessor, including
 - EventProcessor status:
   PAUSED / ACTIVE / ERROR / NEEDS_INVALIDATE / STOPPED / DISABLED.
 - LastSyncedEventId: The last HMS event id which we have synced to.
 - LatestEventId: The event id of the latest event in HMS.

Example output:
[localhost:21050] default> :event_processor('pause');
+--------------------------------------------------------------------------------+
| summary                                                                        |
+--------------------------------------------------------------------------------+
| EventProcessor status: PAUSED. LastSyncedEventId: 34489. LatestEventId: 34489. |
+--------------------------------------------------------------------------------+
Fetched 1 row(s) in 0.01s

If authorization is enabled, only admin users that have ALL privilege on
SERVER can run this command.

Note that there is a restriction in MetastoreEventsProcessor#start(long)
that resuming EventProcessor back to a previous event id is only allowed
when it's not in the ACTIVE state. This patch aims to expose the control
of EventProcessor to the users so MetastoreEventsProcessor is not
changed. We can investigate the restriction and see if we want to relax
it.

Note that resuming EventProcessor at a newer event id can be done on any
states. Admins can use this to manually resolve the lag of HMS event
processing, after they have made sure all (or important) tables are
manually invalidated/refreshed.

A new catalogd RPC, SetEventProcessorStatus, is added for coordinators
to control the status of EventProcessor.

Tests
 - Added e2e tests

Change-Id: I5a19f67264cfe06a1819a22c0c4f0cf174c9b958
Reviewed-on: http://gerrit.cloudera.org:8080/22250
Reviewed-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>

be/src/catalog/catalog-server.cc

@@ -454,6 +454,17 @@ class CatalogServiceThriftIf : public CatalogServiceIf {
     VLOG_RPC << "GetLatestCompactions(): response=" << ThriftDebugStringNoThrow(resp);
   }
 
+  void SetEventProcessorStatus(TSetEventProcessorStatusResponse& resp,
+      const TSetEventProcessorStatusRequest& req) override {
+    VLOG_RPC << "SetEventProcessorStatus(): request=" << ThriftDebugString(req);
+    Status status = AcceptRequest(req.protocol_version);
+    if (status.ok()) {
+      status = catalog_server_->catalog()->SetEventProcessorStatus(req, &resp);
+    }
+    if (!status.ok()) LOG(ERROR) << status.GetDetail();
+    VLOG_RPC << "SetEventProcessorStatus(): response=" << ThriftDebugStringNoThrow(resp);
+  }
+
  private:
   CatalogServer* catalog_server_;
   string server_address_;

be/src/catalog/catalog-service-client-wrapper.h

@@ -123,6 +123,14 @@ class CatalogServiceClientWrapper : public CatalogServiceClient {
     *send_done = true;
     recv_GetLatestCompactions(_return);
   }
+
+  void SetEventProcessorStatus(TSetEventProcessorStatusResponse& _return,
+      const TSetEventProcessorStatusRequest& req, bool* send_done) {
+    DCHECK(!*send_done);
+    send_SetEventProcessorStatus(req);
+    *send_done = true;
+    recv_SetEventProcessorStatus(_return);
+  }
 #pragma clang diagnostic pop
 };
 

be/src/catalog/catalog.cc

@@ -76,6 +76,7 @@ Catalog::Catalog() {
     {"getNullPartitionName", "()[B", &get_null_partition_name_id_},
     {"getLatestCompactions", "([B)[B", &get_latest_compactions_id_},
     {"getAllHadoopConfigs", "()[B", &get_hadoop_configs_id_},
+    {"setEventProcessorStatus", "([B)[B", &set_event_processor_status_id_},
   };
 
   JNIEnv* jni_env = JniUtil::GetJNIEnv();
@@ -238,3 +239,8 @@ Status Catalog::GetLatestCompactions(
     const TGetLatestCompactionsRequest& req, TGetLatestCompactionsResponse* resp) {
   return JniUtil::CallJniMethod(catalog_, get_latest_compactions_id_, req, resp);
 }
+
+Status Catalog::SetEventProcessorStatus(
+    const TSetEventProcessorStatusRequest& req, TSetEventProcessorStatusResponse* resp) {
+  return JniUtil::CallJniMethod(catalog_, set_event_processor_status_id_, req, resp);
+}

be/src/catalog/catalog.h

@@ -155,6 +155,10 @@ class Catalog {
   /// Returns all Hadoop configurations in key, value form in result.
   Status GetAllHadoopConfigs(TGetAllHadoopConfigsResponse* result);
 
+  /// Update the status of EventProcessor.
+  Status SetEventProcessorStatus(
+      const TSetEventProcessorStatusRequest& req, TSetEventProcessorStatusResponse* resp);
+
  private:
   jobject catalog_;  // instance of org.apache.impala.service.JniCatalog
   jmethodID update_metastore_id_;  // JniCatalog.updateMetaastore()
@@ -182,6 +186,7 @@ class Catalog {
   jmethodID get_null_partition_name_id_; // JniCatalog.getNullPartitionName()
   jmethodID get_latest_compactions_id_; // JniCatalog.getLatestCompactions()
   jmethodID get_hadoop_configs_id_;  // JniCatalog.getAllHadoopConfigs()
+  jmethodID set_event_processor_status_id_; // JniCatalog.setEventProcessorStatus()
 };
 
 }

be/src/exec/catalog-op-executor.cc

@@ -475,3 +475,19 @@ Status CatalogOpExecutor::GetLatestCompactions(
   RETURN_IF_ERROR(rpc_status.status);
   return Status::OK();
 }
+
+Status CatalogOpExecutor::SetEventProcessorStatus(
+    const TSetEventProcessorStatusRequest& req,
+    TSetEventProcessorStatusResponse* result) {
+  int attempt = 0; // Used for debug action only.
+  CatalogServiceConnection::RpcStatus rpc_status =
+      CatalogServiceConnection::DoRpcWithRetry(env_->catalogd_client_cache(),
+          *ExecEnv::GetInstance()->GetCatalogdAddress(),
+          &CatalogServiceClientWrapper::SetEventProcessorStatus, req,
+          FLAGS_catalog_client_connection_num_retries,
+          FLAGS_catalog_client_rpc_retry_interval_ms,
+          [&attempt]() { return CatalogRpcDebugFn(&attempt); }, result);
+  RETURN_IF_ERROR(rpc_status.status);
+  if (result->status.status_code != TErrorCode::OK) return Status(result->status);
+  return Status::OK();
+}

be/src/exec/catalog-op-executor.h

@@ -88,6 +88,10 @@ class CatalogOpExecutor {
   Status GetLatestCompactions(
       const TGetLatestCompactionsRequest& req, TGetLatestCompactionsResponse* result);
 
+  /// Makes an RPC to the catalog server to update the status of EventProcessor.
+  Status SetEventProcessorStatus(const TSetEventProcessorStatusRequest& req,
+      TSetEventProcessorStatusResponse* result);
+
   /// Set in Exec(), returns a pointer to the TDdlExecResponse of the DDL execution.
   /// If called before Exec(), this will return NULL. Only set if the
   /// TCatalogOpType is DDL.

be/src/service/client-request-state.cc

@@ -349,8 +349,13 @@ Status ClientRequestState::Exec() {
       break;
     }
     case TStmtType::ADMIN_FN:
-      DCHECK(exec_req.admin_request.type == TAdminRequestType::SHUTDOWN);
-      RETURN_IF_ERROR(ExecShutdownRequest());
+      if (exec_req.admin_request.type == TAdminRequestType::SHUTDOWN) {
+        RETURN_IF_ERROR(ExecShutdownRequest());
+      } else if (exec_req.admin_request.type == TAdminRequestType::EVENT_PROCESSOR) {
+        RETURN_IF_ERROR(ExecEventProcessorCmd());
+      } else {
+        DCHECK(false);
+      }
       break;
     case TStmtType::CONVERT:
       DCHECK(exec_req.__isset.convert_table_request);
@@ -1094,6 +1099,24 @@ Status ClientRequestState::ExecShutdownRequest() {
   return Status::OK();
 }
 
+Status ClientRequestState::ExecEventProcessorCmd() {
+  catalog_op_executor_.reset(
+      new CatalogOpExecutor(ExecEnv::GetInstance(), frontend_, server_profile_));
+  const TEventProcessorCmdParams& params =
+      exec_request().admin_request.event_processor_cmd_params;
+  TSetEventProcessorStatusRequest request;
+  TSetEventProcessorStatusResponse response;
+  request.__set_params(params);
+  request.__set_header(GetCatalogServiceRequestHeader());
+  Status rpc_status = catalog_op_executor_->SetEventProcessorStatus(request, &response);
+  if (!rpc_status.ok()) {
+    VLOG_QUERY << "SetEventProcessorStatus failed: " << rpc_status.msg().msg();
+    return rpc_status;
+  }
+  SetResultSet({response.info});
+  return Status::OK();
+}
+
 void ClientRequestState::Finalize(const Status* cause) {
   UnRegisterCompletedRPCs();
   Cancel(cause, /*wait_until_finalized=*/true);

be/src/service/client-request-state.h

@@ -873,6 +873,9 @@ class ClientRequestState {
   /// Executes a shut down request.
   Status ExecShutdownRequest() WARN_UNUSED_RESULT;
 
+  /// Executes a command on EventProcessor.
+  Status ExecEventProcessorCmd() WARN_UNUSED_RESULT;
+
   /// Core logic of Wait(). Does not update operation_state_/query_status_.
   Status WaitInternal() WARN_UNUSED_RESULT;
 

common/thrift/CatalogService.thrift

@@ -754,6 +754,24 @@ struct TUpdateTableUsageResponse {
   1: optional Status.TStatus status
 }
 
+struct TEventProcessorCmdParams {
+  // See allowed actions in Java enum MetastoreEventsProcessor.EventProcessorCmdType.
+  // Use string type instead of enum to avoid incompatible thrift changes in the future.
+  1: required string action
+  2: optional i64 event_id
+}
+
+struct TSetEventProcessorStatusRequest {
+  1: required CatalogServiceVersion protocol_version = CatalogServiceVersion.V2
+  2: optional TCatalogServiceRequestHeader header
+  3: required TEventProcessorCmdParams params
+}
+
+struct TSetEventProcessorStatusResponse {
+  1: required Status.TStatus status
+  2: optional string info
+}
+
 // The CatalogService API
 service CatalogService {
   // Executes a DDL request and returns details on the result of the operation.
@@ -794,4 +812,8 @@ service CatalogService {
 
   // Gets the latest compactions.
   TGetLatestCompactionsResponse GetLatestCompactions(1: TGetLatestCompactionsRequest req);
+
+  // Update the status of EventProcessor.
+  TSetEventProcessorStatusResponse SetEventProcessorStatus(
+      1: TSetEventProcessorStatusRequest req);
 }

common/thrift/Frontend.thrift

@@ -556,6 +556,7 @@ struct TShutdownParams {
 // The type of administrative function to be executed.
 enum TAdminRequestType {
   SHUTDOWN = 0
+  EVENT_PROCESSOR = 1
 }
 
 // Parameters for administrative function statement. This is essentially a tagged union
@@ -565,6 +566,7 @@ struct TAdminRequest {
 
   // The below member corresponding to 'type' should be set.
   2: optional TShutdownParams shutdown_params
+  3: optional CatalogService.TEventProcessorCmdParams event_processor_cmd_params
 }
 
 // HiveServer2 Metadata operations (JniFrontend.hiveServer2MetadataOperation)

fe/src/main/java/org/apache/impala/analysis/AdminFnStmt.java

@@ -26,6 +26,7 @@ import org.apache.impala.common.InternalException;
 import org.apache.impala.common.Pair;
 import org.apache.impala.thrift.TAdminRequest;
 import org.apache.impala.thrift.TAdminRequestType;
+import org.apache.impala.thrift.TEventProcessorCmdParams;
 import org.apache.impala.thrift.TNetworkAddress;
 import org.apache.impala.thrift.TShutdownParams;
 
@@ -36,8 +37,8 @@ import com.google.common.base.Preconditions;
  * Represents an administrative function call, e.g. ": shutdown('hostname:123')".
  *
  * This "admin statement" framework provides a way to expand the set of supported admin
- * statements without modifying the SQL grammar. For now, the only supported function is
- * shutdown(), so the logic in here is not generic.
+ * statements without modifying the SQL grammar. For now, the only supported functions are
+ * shutdown() and event_processor(), so the logic in here is not generic.
  */
 public class AdminFnStmt extends StatementBase {
   // Name of the function. Validated during analysis.
@@ -46,6 +47,8 @@ public class AdminFnStmt extends StatementBase {
   // Arguments to the function. Always non-null.
   private final List<Expr> params_;
 
+  private TAdminRequestType type_;
+
   // Parameters for the shutdown() command.
   // Address of the backend to shut down, If 'backend_' is null, that means the current
   // server. If 'backend_.port' is 0, we assume the backend has the same port as this
@@ -54,6 +57,13 @@ public class AdminFnStmt extends StatementBase {
   // Deadline in seconds. -1 if no deadline specified.
   private long deadlineSecs_;
 
+  // Parameters for the event_processor() command
+  // Currently supported actions: pause, start
+  private String action_;
+  // Event id to start at. Defaults to reusing the last synced event id.
+  // Set to -1 for using the latest event id.
+  private long event_id_ = 0;
+
   public AdminFnStmt(String fnName, List<Expr> params) {
     this.fnName_ = fnName;
     this.params_ = params;
@@ -72,10 +82,17 @@ public class AdminFnStmt extends StatementBase {
 
   public TAdminRequest toThrift() throws InternalException {
     TAdminRequest result = new TAdminRequest();
-    result.type = TAdminRequestType.SHUTDOWN;
-    result.shutdown_params = new TShutdownParams();
-    if (backend_ != null) result.shutdown_params.setBackend(backend_);
-    if (deadlineSecs_ != -1) result.shutdown_params.setDeadline_s(deadlineSecs_);
+    result.type = type_;
+    if (type_ == TAdminRequestType.SHUTDOWN) {
+      result.shutdown_params = new TShutdownParams();
+      if (backend_ != null) result.shutdown_params.setBackend(backend_);
+      if (deadlineSecs_ != -1) result.shutdown_params.setDeadline_s(deadlineSecs_);
+    } else if (type_ == TAdminRequestType.EVENT_PROCESSOR) {
+      result.event_processor_cmd_params = new TEventProcessorCmdParams(action_);
+      if (event_id_ != 0) result.event_processor_cmd_params.setEvent_id(event_id_);
+    } else {
+      Preconditions.checkState(false, "Unsupported TAdminRequest type %s", type_);
+    }
     return result;
   }
 
@@ -83,26 +100,33 @@ public class AdminFnStmt extends StatementBase {
   public void analyze(Analyzer analyzer) throws AnalysisException {
     super.analyze(analyzer);
     for (Expr param : params_) param.analyze(analyzer);
-    // Only shutdown is supported.
-    if (fnName_.toLowerCase().equals("shutdown")) {
+    if (fnName_.equalsIgnoreCase("shutdown")) {
+      type_ = TAdminRequestType.SHUTDOWN;
       analyzeShutdown(analyzer);
+    } else if (fnName_.equalsIgnoreCase("event_processor")) {
+      type_ = TAdminRequestType.EVENT_PROCESSOR;
+      analyzeEventProcessorCmd(analyzer);
     } else {
       throw new AnalysisException("Unknown admin function: " + fnName_);
     }
   }
 
+  private void registerPrivReq(Analyzer analyzer) {
+    AuthorizationConfig authzConfig = analyzer.getAuthzConfig();
+    if (authzConfig.isEnabled()) {
+      // Only admins (i.e. user with ALL privilege on server) can execute admin functions.
+      String authzServer = authzConfig.getServerName();
+      Preconditions.checkNotNull(authzServer);
+      analyzer.registerPrivReq(builder -> builder.onServer(authzServer).all().build());
+    }
+  }
+
   /**
    * Supports optionally specifying the backend and the deadline: either shutdown(),
    * shutdown('host:port'), shutdown(deadline), shutdown('host:port', deadline).
    */
   private void analyzeShutdown(Analyzer analyzer) throws AnalysisException {
-    AuthorizationConfig authzConfig = analyzer.getAuthzConfig();
-    if (authzConfig.isEnabled()) {
-      // Only admins (i.e. user with ALL privilege on server) can execute admin functions.
-      String authzServer = authzConfig.getServerName();
-      Preconditions.checkNotNull(authzServer);
-      analyzer.registerPrivReq(builder -> builder.onServer(authzServer).all().build());
-    }
+    registerPrivReq(analyzer);
 
     // TODO: this parsing and type checking logic is specific to the command, similar to
     // handling of other top-level commands. If we add a lot more of these functions we
@@ -162,4 +186,19 @@ public class AdminFnStmt extends StatementBase {
     }
     return result;
   }
+
+  private void analyzeEventProcessorCmd(Analyzer analyzer) throws AnalysisException {
+    registerPrivReq(analyzer);
+
+    if (params_.isEmpty() || params_.size() > 2) {
+      throw new AnalysisException("event_processor() takes 1 or 2 arguments: " + toSql());
+    }
+    if (!(params_.get(0) instanceof StringLiteral)) {
+      throw new AnalysisException("First argument of event_processor() should be STRING");
+    }
+    action_ = ((StringLiteral)params_.get(0)).getStringValue().toUpperCase();
+    if (params_.size() > 1) {
+      event_id_ = params_.get(1).evalToInteger(analyzer, "event_id");
+    }
+  }
 }

fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java

@@ -78,6 +78,7 @@ import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventFactory;
 import org.apache.impala.catalog.events.MetastoreEventsProcessor;
 import org.apache.impala.catalog.events.MetastoreEventsProcessor.EventProcessorStatus;
 import org.apache.impala.catalog.events.MetastoreNotificationFetchException;
+import org.apache.impala.catalog.events.NoOpEventProcessor;
 import org.apache.impala.catalog.events.SelfEventContext;
 import org.apache.impala.catalog.metastore.CatalogHmsUtils;
 import org.apache.impala.catalog.monitor.CatalogMonitor;
@@ -107,6 +108,8 @@ import org.apache.impala.thrift.TCatalogObjectType;
 import org.apache.impala.thrift.TCatalogUpdateResult;
 import org.apache.impala.thrift.TDataSource;
 import org.apache.impala.thrift.TDatabase;
+import org.apache.impala.thrift.TErrorCode;
+import org.apache.impala.thrift.TEventProcessorCmdParams;
 import org.apache.impala.thrift.TEventProcessorMetrics;
 import org.apache.impala.thrift.TEventProcessorMetricsSummaryResponse;
 import org.apache.impala.thrift.TFunction;
@@ -126,6 +129,8 @@ import org.apache.impala.thrift.TPartitionStats;
 import org.apache.impala.thrift.TPrincipalType;
 import org.apache.impala.thrift.TPrivilege;
 import org.apache.impala.thrift.TResetMetadataRequest;
+import org.apache.impala.thrift.TSetEventProcessorStatusResponse;
+import org.apache.impala.thrift.TStatus;
 import org.apache.impala.thrift.TSystemTableName;
 import org.apache.impala.thrift.TTable;
 import org.apache.impala.thrift.TTableName;
@@ -3736,6 +3741,82 @@ public class CatalogServiceCatalog extends Catalog {
     return metastoreEventProcessor_.getEventProcessorSummary();
   }
 
+  public TSetEventProcessorStatusResponse setEventProcessorStatus(
+      TEventProcessorCmdParams params) {
+    TSetEventProcessorStatusResponse resp = new TSetEventProcessorStatusResponse();
+    if (metastoreEventProcessor_ instanceof NoOpEventProcessor) {
+      resp.setStatus(new TStatus(TErrorCode.GENERAL,
+          Lists.newArrayList("EventProcessor is disabled")));
+      return resp;
+    }
+    MetastoreEventsProcessor.EventProcessorCmdType cmdType;
+    try {
+      cmdType = MetastoreEventsProcessor.EventProcessorCmdType.valueOf(params.action);
+    } catch (IllegalArgumentException e) {
+      String msg = "Unknown command: " + params.action + ". Supported commands: " +
+          Arrays.stream(MetastoreEventsProcessor.EventProcessorCmdType.values())
+              .map(Enum::name)
+              .collect(Collectors.joining(", "));
+      resp.setStatus(new TStatus(TErrorCode.GENERAL, Lists.newArrayList(msg)));
+      return resp;
+    }
+    MetastoreEventsProcessor ep = (MetastoreEventsProcessor) metastoreEventProcessor_;
+    StringBuilder info = new StringBuilder();
+    if (cmdType == MetastoreEventsProcessor.EventProcessorCmdType.PAUSE) {
+      ep.pause();
+    } else if (cmdType == MetastoreEventsProcessor.EventProcessorCmdType.START) {
+      if (!startEventProcessorHelper(params, ep, resp, info)) {
+        // 'resp' is updated in startEventProcessorHelper() for errors.
+        return resp;
+      }
+    }
+    info.append(String.format("EventProcessor status: %s. LastSyncedEventId: %d. " +
+        "LatestEventId: %d.",
+        ep.getStatus(), ep.getLastSyncedEventId(), ep.getLatestEventId()));
+    resp.setStatus(new TStatus(TErrorCode.OK, Collections.emptyList()));
+    resp.setInfo(info.toString());
+    return resp;
+  }
+
+  private boolean startEventProcessorHelper(TEventProcessorCmdParams params,
+      MetastoreEventsProcessor ep, TSetEventProcessorStatusResponse resp,
+      StringBuilder warnings) {
+    if (!params.isSetEvent_id()) {
+      if (ep.getStatus() != EventProcessorStatus.ACTIVE) {
+        ep.start(ep.getLastSyncedEventId());
+      }
+      return true;
+    }
+    if (params.getEvent_id() < -1) {
+      String msg = "Illegal event id " + params.getEvent_id() + ". Should be >= -1";
+      resp.setStatus(new TStatus(TErrorCode.GENERAL, Lists.newArrayList(msg)));
+      return false;
+    }
+    if (params.getEvent_id() == -1) {
+      ep.start(ep.getLatestEventId());
+      return true;
+    }
+    long lastSyncedEventId = ep.getLastSyncedEventId();
+    if (params.getEvent_id() < lastSyncedEventId
+        && ep.getStatus() == EventProcessorStatus.ACTIVE) {
+      String err = "EventProcessor is active. Failed to set last synced event id " +
+          "from " + lastSyncedEventId + " back to " + params.getEvent_id() +
+          ". Please pause EventProcessor first.";
+      resp.setStatus(new TStatus(TErrorCode.GENERAL, Lists.newArrayList(err)));
+      return false;
+    }
+    long latestEventId = ep.getLatestEventId();
+    if (params.getEvent_id() > latestEventId) {
+      String warning = String.format("Target event id %d is larger than the latest " +
+          "event id %d. Some future events will be skipped.",
+          params.getEvent_id(), latestEventId);
+      LOG.warn(warning);
+      warnings.append(warning).append("\n");
+    }
+    ep.start(params.getEvent_id());
+    return true;
+  }
+
   /**
    * Retrieves the stored metrics of the specified table and returns a pretty-printed
    * string representation. Throws an exception if table metrics were not available

fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java

@@ -607,6 +607,15 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
     DISABLED // event processor is not configured to run
   }
 
+  // supported commands for :event_processor(), e.g.
+  //   :event_processor('pause');
+  //   :event_processor('start', -1);
+  public enum EventProcessorCmdType {
+    PAUSE,
+    START,
+    STATUS,
+  }
+
   // current status of this event processor
   private EventProcessorStatus eventProcessorStatus_ = EventProcessorStatus.STOPPED;
 

fe/src/main/java/org/apache/impala/service/JniCatalog.java

@@ -64,6 +64,7 @@ import org.apache.impala.thrift.TCatalogObject;
 import org.apache.impala.thrift.TDatabase;
 import org.apache.impala.thrift.TDdlExecRequest;
 import org.apache.impala.thrift.TErrorCode;
+import org.apache.impala.thrift.TEventProcessorCmdParams;
 import org.apache.impala.thrift.TGetCatalogDeltaRequest;
 import org.apache.impala.thrift.TGetCatalogDeltaResponse;
 import org.apache.impala.thrift.TGetCatalogServerMetricsResponse;
@@ -83,6 +84,7 @@ import org.apache.impala.thrift.TGetTablesResult;
 import org.apache.impala.thrift.TLogLevel;
 import org.apache.impala.thrift.TPrioritizeLoadRequest;
 import org.apache.impala.thrift.TResetMetadataRequest;
+import org.apache.impala.thrift.TSetEventProcessorStatusRequest;
 import org.apache.impala.thrift.TStatus;
 import org.apache.impala.thrift.TTableUsage;
 import org.apache.impala.thrift.TUniqueId;
@@ -526,6 +528,19 @@ public class JniCatalog {
         "getEventProcessorSummary", shortDesc, catalog_::getEventProcessorSummary);
   }
 
+  public byte[] setEventProcessorStatus(byte[] thriftParams)
+      throws ImpalaException, TException {
+    TSetEventProcessorStatusRequest request = new TSetEventProcessorStatusRequest();
+    JniUtil.deserializeThrift(protocolFactory_, request, thriftParams);
+    String shortDesc = request.params.action + " event processor";
+    if (request.params.isSetEvent_id()) {
+      shortDesc += " at event id " + request.params.event_id;
+    }
+    return execAndSerialize(
+        "setEventProcessorStatus", shortDesc,
+        () -> catalog_.setEventProcessorStatus(request.params));
+  }
+
   public void updateTableUsage(byte[] req) throws ImpalaException, TException {
     TUpdateTableUsageRequest thriftReq = new TUpdateTableUsageRequest();
     JniUtil.deserializeThrift(protocolFactory_, thriftReq, req);

tests/metadata/test_event_processing.py

@@ -17,6 +17,8 @@
 from __future__ import absolute_import, division, print_function
 from subprocess import check_call
 import pytest
+import re
+import time
 
 from tests.common.impala_cluster import ImpalaCluster
 from tests.common.impala_test_suite import ImpalaTestSuite
@@ -205,3 +207,83 @@ class TestEventProcessing(ImpalaTestSuite):
           break
         return row[1].rstrip()
     return None
+
+  def _exec_and_check_ep_cmd(self, cmd, expected_status):
+    cmd_output = self.execute_query(cmd).get_data()
+    match = re.search(
+      r"EventProcessor status: %s. LastSyncedEventId: \d+. LatestEventId: \d+." %
+      expected_status,
+      cmd_output)
+    assert match, cmd_output
+    assert EventProcessorUtils.get_event_processor_status() == expected_status
+    return cmd_output
+
+  @pytest.mark.execute_serially
+  def test_event_processor_cmds(self, unique_database):
+    ###########################################################################
+    # 1. Test normal PAUSE and RESUME. Also check the STATUS command.
+    self._exec_and_check_ep_cmd(":event_processor('pause')", "PAUSED")
+    self._exec_and_check_ep_cmd(":event_processor('status')", "PAUSED")
+    self._exec_and_check_ep_cmd(":event_processor('start')", "ACTIVE")
+    self._exec_and_check_ep_cmd(":event_processor('status')", "ACTIVE")
+
+    # Make sure the CREATE_DATABASE event for 'unique_database' is processed
+    EventProcessorUtils.wait_for_event_processing(self)
+
+    ###########################################################################
+    # 2. Test failure of restarting at an older event id when status is ACTIVE
+    last_synced_event_id = EventProcessorUtils.get_last_synced_event_id()
+    e = self.execute_query_expect_failure(
+        self.client, ":event_processor('start', %d)" % (last_synced_event_id / 2))
+    assert "EventProcessor is active. Failed to set last synced event id from " +\
+        str(last_synced_event_id) + " back to " + str(int(last_synced_event_id / 2)) +\
+        ". Please pause EventProcessor first." in str(e)
+
+    ###########################################################################
+    # 3. Test restarting to the latest event id
+    self._exec_and_check_ep_cmd(":event_processor('pause')", "PAUSED")
+    # Create some HMS events
+    for i in range(3):
+      self.run_stmt_in_hive("create table %s.tbl_%d(i int)" % (unique_database, i))
+    latest_event_id = EventProcessorUtils.get_current_notification_id(self.hive_client)
+    # Wait some time for EP to update its latest event id
+    time.sleep(2)
+    # Restart to the latest event id
+    self._exec_and_check_ep_cmd(":event_processor('start', -1)", "ACTIVE")
+    assert EventProcessorUtils.get_last_synced_event_id() == latest_event_id
+    # Verify the new events are skipped so Impala queries should fail
+    for i in range(3):
+      self.execute_query_expect_failure(
+          self.client, "describe %s.tbl_%d" % (unique_database, i))
+
+    ###########################################################################
+    # 4. Test setting back the last synced event id after pausing EP
+    self._exec_and_check_ep_cmd(":event_processor('pause')", "PAUSED")
+    # Restart to the previous last synced event id to process the missing HMS events
+    self._exec_and_check_ep_cmd(
+        ":event_processor('start', %d)" % last_synced_event_id, "ACTIVE")
+    EventProcessorUtils.wait_for_event_processing(self)
+    # Tables should be visible now
+    for i in range(3):
+      self.execute_query_expect_success(
+          self.client, "describe %s.tbl_%d" % (unique_database, i))
+
+    ###########################################################################
+    # 5. Test unknown commands
+    e = self.execute_query_expect_failure(self.client, ":event_processor('bad_cmd')")
+    assert "Unknown command: BAD_CMD. Supported commands: PAUSE, START, STATUS" in str(e)
+
+    ###########################################################################
+    # 6. Test illegal event id
+    e = self.execute_query_expect_failure(self.client, ":event_processor('start', -2)")
+    assert "Illegal event id -2. Should be >= -1" in str(e)
+
+    ###########################################################################
+    # 7. Test restarting on a future event id
+    cmd_output = self._exec_and_check_ep_cmd(
+        ":event_processor('start', %d)" % (latest_event_id + 2), "ACTIVE")
+    warning = ("Target event id %d is larger than the latest event id %d. Some future "
+               "events will be skipped.") % (latest_event_id + 2, latest_event_id)
+    assert warning in cmd_output
+    # The cleanup method will drop 'unique_database' and tables in it, which generates
+    # more than 2 self-events. It's OK for EP to skip them.