address comments 2

apache · Sep 12, 2023 · c465389 · c465389
1 parent 9d9cf4e
commit c465389
Show file tree

Hide file tree

Showing 12 changed files with 43 additions and 28 deletions.
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
@@ -70,10 +70,10 @@
 import org.apache.pinot.core.util.ListenerConfigUtil;
 import org.apache.pinot.spi.accounting.ThreadResourceUsageProvider;
 import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListener;
+import org.apache.pinot.spi.eventlistener.query.PinotBrokerQueryEventListenerUtils;
 import org.apache.pinot.spi.metrics.PinotMetricUtils;
 import org.apache.pinot.spi.metrics.PinotMetricsRegistry;
-import org.apache.pinot.spi.queryeventlistener.BrokerQueryEventListener;
-import org.apache.pinot.spi.queryeventlistener.PinotBrokerQueryEventListenerUtils;
 import org.apache.pinot.spi.services.ServiceRole;
 import org.apache.pinot.spi.services.ServiceStartable;
 import org.apache.pinot.spi.trace.Tracing;

diff --git a/...broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java b/...broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
@@ -89,9 +89,9 @@
 import org.apache.pinot.spi.config.table.TableType;
 import org.apache.pinot.spi.data.Schema;
 import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventInfo;
+import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListener;
 import org.apache.pinot.spi.exception.BadQueryRequestException;
-import org.apache.pinot.spi.queryeventlistener.BrokerQueryEventInfo;
-import org.apache.pinot.spi.queryeventlistener.BrokerQueryEventListener;
 import org.apache.pinot.spi.trace.RequestContext;
 import org.apache.pinot.spi.trace.Tracing;
 import org.apache.pinot.spi.utils.BytesUtils;
@@ -246,6 +246,9 @@ public BrokerResponse handleRequest(JsonNode request, @Nullable SqlNodeAndOption
       throws Exception {
     requestContext.setRequestArrivalTimeMillis(System.currentTimeMillis());
 
+    long requestId = _brokerIdGenerator.get();
+    requestContext.setRequestId(requestId);
+
     // First-stage access control to prevent unauthenticated requests from using up resources. Secondary table-level
     // check comes later.
     boolean hasAccess = _accessControlFactory.create().hasAccess(requesterIdentity);
@@ -256,10 +259,9 @@ public BrokerResponse handleRequest(JsonNode request, @Nullable SqlNodeAndOption
       throw new WebApplicationException("Permission denied", Response.Status.FORBIDDEN);
     }
 
-    long requestId = _brokerIdGenerator.get();
-    requestContext.setRequestId(requestId);
     JsonNode sql = request.get(Broker.Request.SQL);
     if (sql == null) {
+      requestContext.setErrorCode(QueryException.BROKER_REQUEST_SEND_ERROR_CODE);
       _brokerQueryEventListener.onQueryCompletion(new BrokerQueryEventInfo(requestContext));
       throw new BadQueryRequestException("Failed to find 'sql' in the request: " + request);
     }

diff --git a/...broker/src/main/java/org/apache/pinot/broker/requesthandler/GrpcBrokerRequestHandler.java b/...broker/src/main/java/org/apache/pinot/broker/requesthandler/GrpcBrokerRequestHandler.java
@@ -42,7 +42,7 @@
 import org.apache.pinot.core.transport.ServerRoutingInstance;
 import org.apache.pinot.spi.config.table.TableType;
 import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.queryeventlistener.BrokerQueryEventListener;
+import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListener;
 import org.apache.pinot.spi.trace.RequestContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

diff --git a/.../src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java b/.../src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -62,7 +62,7 @@
 import org.apache.pinot.query.type.TypeFactory;
 import org.apache.pinot.query.type.TypeSystem;
 import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.queryeventlistener.BrokerQueryEventListener;
+import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListener;
 import org.apache.pinot.spi.trace.RequestContext;
 import org.apache.pinot.spi.utils.CommonConstants;
 import org.apache.pinot.spi.utils.builder.TableNameBuilder;

diff --git a/...ain/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java b/...ain/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
@@ -50,7 +50,7 @@
 import org.apache.pinot.core.transport.ServerRoutingInstance;
 import org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
 import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.queryeventlistener.BrokerQueryEventListener;
+import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListener;
 import org.apache.pinot.spi.trace.RequestContext;
 import org.apache.pinot.spi.utils.CommonConstants;
 import org.apache.pinot.spi.utils.builder.TableNameBuilder;

diff --git a/...er/src/test/java/org/apache/pinot/broker/requesthandler/LiteralOnlyBrokerRequestTest.java b/...er/src/test/java/org/apache/pinot/broker/requesthandler/LiteralOnlyBrokerRequestTest.java
@@ -31,8 +31,8 @@
 import org.apache.pinot.common.utils.DataSchema;
 import org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
 import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.eventlistener.query.PinotBrokerQueryEventListenerUtils;
 import org.apache.pinot.spi.metrics.PinotMetricUtils;
-import org.apache.pinot.spi.queryeventlistener.PinotBrokerQueryEventListenerUtils;
 import org.apache.pinot.spi.trace.RequestContext;
 import org.apache.pinot.spi.trace.Tracing;
 import org.apache.pinot.spi.utils.BytesUtils;

diff --git a/.../test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java b/.../test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
@@ -29,7 +29,7 @@
 import org.apache.pinot.common.config.provider.TableCache;
 import org.apache.pinot.common.metrics.BrokerMetrics;
 import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.queryeventlistener.PinotBrokerQueryEventListenerUtils;
+import org.apache.pinot.spi.eventlistener.query.PinotBrokerQueryEventListenerUtils;
 import org.apache.pinot.spi.trace.DefaultRequestContext;
 import org.apache.pinot.spi.trace.RequestContext;
 import org.apache.pinot.spi.utils.CommonConstants;

diff --git a/...ryeventlistener/BrokerQueryEventInfo.java → ...tlistener/query/BrokerQueryEventInfo.java b/...ryeventlistener/BrokerQueryEventInfo.java → ...tlistener/query/BrokerQueryEventInfo.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.spi.queryeventlistener;
+package org.apache.pinot.spi.eventlistener.query;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -25,15 +25,15 @@
 
 
 public class BrokerQueryEventInfo {
-  private String _requestId;
-  private String _brokerId;
-  private String _query;
-  private String _queryStatus;
-  private String _failureJson;
-  private int _errorCode;
-
-  private long _requestArrivalTimeMillis;
-  private int _numServersQueried;
+  private String _requestId = "";
+  private String _brokerId = "";
+  private String _query = "";
+  private String _queryStatus = "";
+  private String _failureJson = "";
+  private int _errorCode = -1;
+
+  private long _requestArrivalTimeMillis = 0L;
+  private int _numServersQueried = 0;
   private int _numServersResponded = 0;
   private long _numDocsScanned = 0L;
   private long _numEntriesScannedInFilter = 0L;
@@ -67,8 +67,8 @@ public class BrokerQueryEventInfo {
   private long _explainPlanNumMatchAllFilterSegments = 0L;
   private int _numRowsResultSet = 0;
   private List<String> _tableNames = new ArrayList<>();
-  private String _offlineServerTenant;
-  private String _realtimeServerTenant;
+  private String _offlineServerTenant = "";
+  private String _realtimeServerTenant = "";
 
   public BrokerQueryEventInfo() {
   }

diff --git a/...entlistener/BrokerQueryEventListener.java → ...tener/query/BrokerQueryEventListener.java b/...entlistener/BrokerQueryEventListener.java → ...tener/query/BrokerQueryEventListener.java
@@ -16,9 +16,13 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.spi.queryeventlistener;
+package org.apache.pinot.spi.eventlistener.query;
+
+import org.apache.pinot.spi.env.PinotConfiguration;
+
 
 public interface BrokerQueryEventListener {
 
+  void init(PinotConfiguration eventListenerConfiguration);
   void onQueryCompletion(BrokerQueryEventInfo brokerQueryEventInfo);
 }
diff --git a/...istener/NoOpBrokerQueryEventListener.java → ...r/query/NoOpBrokerQueryEventListener.java b/...istener/NoOpBrokerQueryEventListener.java → ...r/query/NoOpBrokerQueryEventListener.java
@@ -16,10 +16,18 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.spi.queryeventlistener;
+package org.apache.pinot.spi.eventlistener.query;
+
+import org.apache.pinot.spi.env.PinotConfiguration;
+
 
 public class NoOpBrokerQueryEventListener implements BrokerQueryEventListener {
 
+  @Override
+  public void init(PinotConfiguration eventListenerConfiguration) {
+    // Not implemented method
+  }
+
   @Override
   public void onQueryCompletion(BrokerQueryEventInfo brokerQueryEventInfo) {
     // Not implemented method

diff --git a/...r/PinotBrokerQueryEventListenerUtils.java → ...y/PinotBrokerQueryEventListenerUtils.java b/...r/PinotBrokerQueryEventListenerUtils.java → ...y/PinotBrokerQueryEventListenerUtils.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.spi.queryeventlistener;
+package org.apache.pinot.spi.eventlistener.query;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -67,6 +67,7 @@ private static void initializeBrokerQueryEventListener(PinotConfiguration eventL
     clazzFound.ifPresent(clazz -> {
       try {
         BrokerQueryEventListener brokerQueryEventListener = (BrokerQueryEventListener) clazz.newInstance();
+        brokerQueryEventListener.init(eventListenerConfiguration);
         registerBrokerEventListener(brokerQueryEventListener);
       } catch (Exception e) {
         LOGGER.error("Caught exception while initializing event listener registry: {}, skipping it", clazz, e);
@@ -78,7 +79,7 @@ private static void initializeBrokerQueryEventListener(PinotConfiguration eventL
   }
 
   /**
-   * Registers an broker event listener.
+   * Registers a broker event listener.
    */
   private static void registerBrokerEventListener(BrokerQueryEventListener brokerQueryEventListener) {
     LOGGER.info("Registering broker event listener : {}", brokerQueryEventListener.getClass().getName());

diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -46,7 +46,7 @@ private CommonConstants() {
   public static final String DEFAULT_METRICS_FACTORY_CLASS_NAME =
       "org.apache.pinot.plugin.metrics.yammer.YammerMetricsFactory";
   public static final String DEFAULT_BROKER_EVENT_LISTENER_CLASS_NAME =
-      "org.apache.pinot.spi.queryeventlistener.NoOpBrokerQueryEventListener";
+      "org.apache.pinot.spi.eventlistener.query.NoOpBrokerQueryEventListener";
 
   public static final String SWAGGER_AUTHORIZATION_KEY = "oauth";
   public static final String CONFIG_OF_SWAGGER_RESOURCES_PATH = "META-INF/resources/webjars/swagger-ui/5.1.0/";