ctubbsii commented on a change in pull request #2340:
URL: https://github.com/apache/accumulo/pull/2340#discussion_r741297197
##########
File path:
core/src/main/java/org/apache/accumulo/core/client/AccumuloClient.java
##########
@@ -338,6 +338,13 @@ ConditionalWriter createConditionalWriter(String
tableName, ConditionalWriterCon
@Override
void close();
+ /**
+ * Sets a user-defined ClientThreadPools implementation
+ *
+ * @param impl
Review comment:
javadoc quality: remove tag if no description, or add description
##########
File path:
core/src/main/java/org/apache/accumulo/core/client/ClientThreadPools.java
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.accumulo.core.client;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.accumulo.core.clientImpl.ClientContext;
+
+public interface ClientThreadPools {
+
+ /**
+ * return a shared scheduled executor for trivial tasks
+ *
+ * @param ctx
+ * @return ScheduledThreadPoolExecutor
+ */
+ ScheduledThreadPoolExecutor getSharedScheduledExecutor(ClientContext
context);
+
+ /**
+ * ThreadPoolExecutor that runs bulk import tasks
+ *
+ * @param ctx
+ * @param numThreads
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBulkImportThreadPool(ClientContext ctx, int
numThreads);
+
+ /**
+ * ThreadPoolExecutor that runs tasks to contact Compactors to get running
compaction information
+ *
+ * @param numThreads
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getExternalCompactionActiveCompactionsPool(ClientContext
ctx, int numThreads);
+
+ /**
+ * ThreadPoolExecutor used for fetching data from the TabletServers
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getScannerReadAheadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor used for adding splits to a table
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getAddSplitsThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor used for fetching data from the TabletServers
+ *
+ * @param ctx
+ * @param numQueryThreads
+ * @param batchReaderInstance
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchReaderThreadPool(ClientContext ctx, int
numQueryThreads,
+ int batchReaderInstance);
+
+ /**
+ * ScheduledThreadPoolExecutor that runs tasks for the BatchWriter to meet
the users latency
+ * goals.
+ *
+ * @param ctx
+ * client context object
+ * @return ScheduledThreadPoolExecutor
+ */
+ ScheduledThreadPoolExecutor
getBatchWriterLatencyTasksThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor that runs the tasks of binning mutations
+ *
+ * @param ctx
+ * client context object
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchWriterBinningThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor that runs the tasks of sending mutations to
TabletServers
+ *
+ * @param ctx
+ * client context object
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchWriterSendThreadPool(ClientContext ctx, int
numSendThreads);
+
+ /**
+ * ThreadPoolExecutor that runs clean up tasks when close is called on the
ConditionalWriter
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getConditionalWriterCleanupTaskThreadPool(ClientContext
ctx);
Review comment:
Having one central location that entagles itself with so many disparate
public API methods seems less than ideal. I think it would be better if each
entry point (batch scanner, batch writer, conditional writer, etc.) added an
option to their existing builders/config to set a thread pool... or more
narrowly, just an uncaught exception handler for the thread pools we manage
internally.
##########
File path:
core/src/main/java/org/apache/accumulo/core/client/ClientThreadPools.java
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.accumulo.core.client;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.accumulo.core.clientImpl.ClientContext;
+
+public interface ClientThreadPools {
+
+ /**
+ * return a shared scheduled executor for trivial tasks
+ *
+ * @param ctx
+ * @return ScheduledThreadPoolExecutor
+ */
+ ScheduledThreadPoolExecutor getSharedScheduledExecutor(ClientContext
context);
Review comment:
ClientContext is not public API, and should not be
##########
File path:
core/src/main/java/org/apache/accumulo/core/util/threads/AccumuloUncaughtExceptionHandler.java
##########
@@ -44,7 +46,10 @@ public void uncaughtException(Thread t, Throwable e) {
// If e == OutOfMemoryError, then it's probably that another Error
might be
// thrown when trying to print to System.err.
} finally {
- Runtime.getRuntime().halt(-1);
+ Mode m = SingletonManager.getMode();
+ if (m != null && m.equals(Mode.SERVER)) {
+ Runtime.getRuntime().halt(-1);
+ }
Review comment:
Rather than making this handler behave differently in some
circumstances, why not make it so a different handler can be configured instead?
##########
File path:
core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java
##########
@@ -229,6 +228,7 @@ private Path checkPath(FileSystem fs, String dir) throws
IOException, AccumuloEx
}
@Override
+ @Deprecated(since = "2.1.0")
public ImportMappingOptions executor(Executor service) {
Review comment:
It's not clear. Why is this option being deprecated in the bulk import
builder?
##########
File path:
core/src/main/java/org/apache/accumulo/core/client/ClientThreadPools.java
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.accumulo.core.client;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.accumulo.core.clientImpl.ClientContext;
+
+public interface ClientThreadPools {
+
+ /**
+ * return a shared scheduled executor for trivial tasks
+ *
+ * @param ctx
+ * @return ScheduledThreadPoolExecutor
+ */
+ ScheduledThreadPoolExecutor getSharedScheduledExecutor(ClientContext
context);
+
+ /**
+ * ThreadPoolExecutor that runs bulk import tasks
+ *
+ * @param ctx
+ * @param numThreads
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBulkImportThreadPool(ClientContext ctx, int
numThreads);
+
+ /**
+ * ThreadPoolExecutor that runs tasks to contact Compactors to get running
compaction information
+ *
+ * @param numThreads
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getExternalCompactionActiveCompactionsPool(ClientContext
ctx, int numThreads);
+
+ /**
+ * ThreadPoolExecutor used for fetching data from the TabletServers
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getScannerReadAheadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor used for adding splits to a table
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getAddSplitsThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor used for fetching data from the TabletServers
+ *
+ * @param ctx
+ * @param numQueryThreads
+ * @param batchReaderInstance
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchReaderThreadPool(ClientContext ctx, int
numQueryThreads,
+ int batchReaderInstance);
+
+ /**
+ * ScheduledThreadPoolExecutor that runs tasks for the BatchWriter to meet
the users latency
+ * goals.
+ *
+ * @param ctx
+ * client context object
+ * @return ScheduledThreadPoolExecutor
+ */
+ ScheduledThreadPoolExecutor
getBatchWriterLatencyTasksThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor that runs the tasks of binning mutations
+ *
+ * @param ctx
+ * client context object
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchWriterBinningThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor that runs the tasks of sending mutations to
TabletServers
+ *
+ * @param ctx
+ * client context object
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchWriterSendThreadPool(ClientContext ctx, int
numSendThreads);
+
+ /**
+ * ThreadPoolExecutor that runs clean up tasks when close is called on the
ConditionalWriter
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getConditionalWriterCleanupTaskThreadPool(ClientContext
ctx);
Review comment:
We did try to make those builders/config extensible, so they wouldn't
have to modify anything unless they were taking advantage of the new feature...
However, I think I have a better idea anyway.
I noticed that this class is essentially a ThreadPoolExecutorFactory for
either `ThreadPoolExecutor` or `ScheduledThreadPoolExecutor`. Instead of having
so many disparate methods, we can make the interface substantially simpler, and
still support all the different uses, by passing in some kind of context/scope,
as in:
```java
// this name is bad, but illustrates the idea; this could even be a
String, if we don't want to constrain it
enum TheadPoolScope {
BATCH_WRITER, BATCH_SCANNER, CONDITIONAL_WRITER;
}
ThreadPoolExecutor getExecutor(ThreadPoolScope scope, ConfigSource conf);
ScheduledThreadPoolExecutor
getScheduledThreadPoolExecutor(ThreadPoolScope scope, ConfigSource conf);
```
##########
File path:
core/src/main/java/org/apache/accumulo/core/client/ClientThreadPools.java
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.accumulo.core.client;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.accumulo.core.clientImpl.ClientContext;
+
+public interface ClientThreadPools {
+
+ /**
+ * return a shared scheduled executor for trivial tasks
+ *
+ * @param ctx
+ * @return ScheduledThreadPoolExecutor
+ */
+ ScheduledThreadPoolExecutor getSharedScheduledExecutor(ClientContext
context);
+
+ /**
+ * ThreadPoolExecutor that runs bulk import tasks
+ *
+ * @param ctx
+ * @param numThreads
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBulkImportThreadPool(ClientContext ctx, int
numThreads);
+
+ /**
+ * ThreadPoolExecutor that runs tasks to contact Compactors to get running
compaction information
+ *
+ * @param numThreads
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getExternalCompactionActiveCompactionsPool(ClientContext
ctx, int numThreads);
+
+ /**
+ * ThreadPoolExecutor used for fetching data from the TabletServers
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getScannerReadAheadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor used for adding splits to a table
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getAddSplitsThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor used for fetching data from the TabletServers
+ *
+ * @param ctx
+ * @param numQueryThreads
+ * @param batchReaderInstance
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchReaderThreadPool(ClientContext ctx, int
numQueryThreads,
+ int batchReaderInstance);
+
+ /**
+ * ScheduledThreadPoolExecutor that runs tasks for the BatchWriter to meet
the users latency
+ * goals.
+ *
+ * @param ctx
+ * client context object
+ * @return ScheduledThreadPoolExecutor
+ */
+ ScheduledThreadPoolExecutor
getBatchWriterLatencyTasksThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor that runs the tasks of binning mutations
+ *
+ * @param ctx
+ * client context object
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchWriterBinningThreadPool(ClientContext ctx);
+
+ /**
+ * ThreadPoolExecutor that runs the tasks of sending mutations to
TabletServers
+ *
+ * @param ctx
+ * client context object
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getBatchWriterSendThreadPool(ClientContext ctx, int
numSendThreads);
+
+ /**
+ * ThreadPoolExecutor that runs clean up tasks when close is called on the
ConditionalWriter
+ *
+ * @param ctx
+ * @return ThreadPoolExecutor
+ */
+ ThreadPoolExecutor getConditionalWriterCleanupTaskThreadPool(ClientContext
ctx);
Review comment:
We did try to make those builders/config extensible, so they wouldn't
have to modify anything unless they were taking advantage of the new feature...
However, I think I have a better idea anyway.
I noticed that this class is essentially a ThreadPoolExecutorFactory for
either `ThreadPoolExecutor` or `ScheduledThreadPoolExecutor`. Instead of having
so many disparate methods, we can make the interface substantially simpler, and
still support all the different uses, by passing in some kind of context/scope,
as in:
```java
// this name is bad, but illustrates the idea; this could even be a
String, if we don't want to constrain it
enum TheadPoolScope {
BATCH_WRITER, BATCH_SCANNER, CONDITIONAL_WRITER;
}
ThreadPoolExecutor getExecutor(ThreadPoolScope scope, ConfigSource conf);
ScheduledThreadPoolExecutor
getScheduledThreadPoolExecutor(ThreadPoolScope scope, ConfigSource conf);
```
The main idea here is to keep the API as free of bloat as possible, but
still be extensible and applicable to all our use cases.
##########
File path:
core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java
##########
@@ -229,6 +228,7 @@ private Path checkPath(FileSystem fs, String dir) throws
IOException, AccumuloEx
}
@Override
+ @Deprecated(since = "2.1.0")
public ImportMappingOptions executor(Executor service) {
Review comment:
Okay, a javadoc deprecated note to point to the new API would be helpful
here
##########
File path:
core/src/main/java/org/apache/accumulo/core/util/threads/AccumuloUncaughtExceptionHandler.java
##########
@@ -44,7 +46,10 @@ public void uncaughtException(Thread t, Throwable e) {
// If e == OutOfMemoryError, then it's probably that another Error
might be
// thrown when trying to print to System.err.
} finally {
- Runtime.getRuntime().halt(-1);
+ Mode m = SingletonManager.getMode();
+ if (m != null && m.equals(Mode.SERVER)) {
+ Runtime.getRuntime().halt(-1);
+ }
Review comment:
Never mind on this. I think the current idea of providing the executor
factory is better.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
