yashmayya commented on code in PR #15072:
URL: https://github.com/apache/pinot/pull/15072#discussion_r1969360467
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestUtils.java:
##########
@@ -40,6 +40,7 @@
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
import org.apache.pinot.common.utils.request.RequestUtils;
+import org.apache.pinot.spi.executor.MdcExecutor;
Review Comment:
This needs to be re-arranged, it's currently causing the spotless plugin
check to fail.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java:
##########
@@ -140,8 +142,25 @@ public synchronized void shutDown() {
@Override
public InstanceResponseBlock execute(ServerQueryRequest queryRequest,
ExecutorService executorService,
@Nullable ResultsBlockStreamer streamer) {
+ MdcExecutor mdcExecutor = new MdcExecutor(executorService) {
Review Comment:
Do we really need to allocate an instance for every query request? Can't we
use a common `ThreadPoolExecutor` implementation with an overridden
`afterExecute` that clears the MDC context instead?
##########
pinot-tools/src/main/resources/log4j2.xml:
##########
@@ -23,7 +23,12 @@
<Properties>
<Property name="LOG_ROOT">logs</Property>
- <Property name="LOG_PATTERN">%d{yyyy/MM/dd HH:mm:ss.SSS} %p [%c{1}] [%t]
%m%n</Property>
+ <!--
+ See
+
https://logging.apache.org/log4j/2.x/manual/pattern-layout.html#converter-not-empty
+ and
https://logging.apache.org/log4j/2.x/manual/pattern-layout.html#converter-thread-context-map
+ -->
Review Comment:
What's `pinot.component.type`?
##########
pinot-spi/src/main/java/org/apache/pinot/spi/executor/DecoratorExecutorService.java:
##########
@@ -0,0 +1,163 @@
+/**
+ * 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.pinot.spi.executor;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+/**
+ * DecoratorExecutorService is an abstract class that provides a way to
decorate an ExecutorService with additional
+ * functionality.
+ *
+ * TODO: Convert this class and its usages into an Executor instead of an
ExecutorService
+ */
+public abstract class DecoratorExecutorService implements ExecutorService {
+ private final ExecutorService _executorService;
+
+ public DecoratorExecutorService(ExecutorService executorService) {
+ _executorService = executorService;
+ }
+
+ protected abstract <T> Callable<T> decorate(Callable<T> task);
+
+ protected abstract Runnable decorate(Runnable task);
+
+ protected <T> Collection<? extends Callable<T>> decorateTasks(Collection<?
extends Callable<T>> tasks) {
+ return tasks.stream().map(this::decorate).collect(Collectors.toList());
+ }
+
+ @Override
+ public void shutdown() {
+ _executorService.shutdown();
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return _executorService.shutdownNow();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return _executorService.isShutdown();
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return _executorService.isTerminated();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit)
+ throws InterruptedException {
+ return _executorService.awaitTermination(timeout, unit);
+ }
+
+ @Override
+ public <T> Future<T> submit(Callable<T> task) {
+ return _executorService.submit(decorate(task));
+ }
+
+ @Override
+ public <T> Future<T> submit(Runnable task, T result) {
+ return _executorService.submit(decorate(task), result);
+ }
+
+ @Override
+ public Future<?> submit(Runnable task) {
+ return _executorService.submit(decorate(task));
+ }
+
+ @Override
+ public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
+ throws InterruptedException {
+ return _executorService.invokeAll(decorateTasks(tasks));
+ }
+
+ @Override
+ public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
+ long timeout, TimeUnit unit)
+ throws InterruptedException {
+ return _executorService.invokeAll(decorateTasks(tasks), timeout, unit);
+ }
+
+ @Override
+ public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
+ throws InterruptedException, ExecutionException {
+ return _executorService.invokeAny(decorateTasks(tasks));
+ }
+
+ @Override
+ public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout,
+ TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ return _executorService.invokeAny(decorateTasks(tasks), timeout, unit);
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ _executorService.execute(decorate(command));
+ }
+
+ public static abstract class BeforeAfter extends DecoratorExecutorService {
Review Comment:
This is unused? Is it just mean to be an example extensible implementation?
##########
pinot-spi/src/main/java/org/apache/pinot/spi/executor/DecoratorExecutorService.java:
##########
@@ -0,0 +1,163 @@
+/**
+ * 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.pinot.spi.executor;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+/**
+ * DecoratorExecutorService is an abstract class that provides a way to
decorate an ExecutorService with additional
+ * functionality.
+ *
+ * TODO: Convert this class and its usages into an Executor instead of an
ExecutorService
+ */
+public abstract class DecoratorExecutorService implements ExecutorService {
+ private final ExecutorService _executorService;
+
+ public DecoratorExecutorService(ExecutorService executorService) {
+ _executorService = executorService;
+ }
+
+ protected abstract <T> Callable<T> decorate(Callable<T> task);
+
+ protected abstract Runnable decorate(Runnable task);
+
+ protected <T> Collection<? extends Callable<T>> decorateTasks(Collection<?
extends Callable<T>> tasks) {
Review Comment:
nit: some Javadocs could be useful here
##########
pinot-spi/src/main/java/org/apache/pinot/spi/trace/LoggerConstants.java:
##########
@@ -0,0 +1,65 @@
+/**
+ * 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.pinot.spi.trace;
+
+import org.slf4j.MDC;
+
+
+public enum LoggerConstants {
+
+ QUERY_ID_KEY("pinot.query.id"),
+ /**
+ * The MSE worker id of the query.
+ */
+ WORKER_ID_KEY("pinot.mse.workerId"),
+ /**
+ * The MSE stage id of the query.
+ */
+ STAGE_ID_KEY("pinot.mse.stageId");
+
+ private final String _key;
+
+ LoggerConstants(String key) {
+ _key = key;
+ }
+
+ public String getKey() {
+ return _key;
+ }
+
+ public void registerOnMdc(String value) {
+ registerOnMdcIfNotSet(value, false);
+ }
+
+ public boolean registerOnMdcIfNotSet(String value, boolean override) {
Review Comment:
Even if the older value is the same as the value we want to register (in
case we already set it at some previous point in the thread's execution) it
doesn't hurt to override the value right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
