[
https://issues.apache.org/jira/browse/PHOENIX-6155?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17233160#comment-17233160
]
ASF GitHub Bot commented on PHOENIX-6155:
-----------------------------------------
ChinmaySKulkarni commented on a change in pull request #960:
URL: https://github.com/apache/phoenix/pull/960#discussion_r524762304
##########
File path: phoenix-core/src/main/java/org/apache/phoenix/schema/task/Task.java
##########
@@ -75,6 +87,55 @@ public Void run() throws Exception {
}
}
+ private static List<Mutation> getMutationsForSystemTaskTable(
+ PhoenixConnection conn, PreparedStatement stmt,
+ boolean accessCheckEnabled) throws IOException {
+ // we need to mutate SYSTEM.TASK with HBase/login user if access is
enabled.
+ if (accessCheckEnabled) {
+ return User.runAsLoginUser(() -> {
+ final RpcCall rpcContext = RpcUtil.getRpcContext();
+ // setting RPC context as null so that user can be reset
+ try {
+ RpcUtil.setRpcContext(null);
+ stmt.execute();
+ // retrieve mutations for SYSTEM.TASK upsert query
+ Iterator<Pair<byte[], List<Mutation>>> iterator =
+ conn.getMutationState().toMutations();
+ List<Mutation> taskMutations = iterator.next().getSecond();
+ // we are expecting conn to be used for single upsert
Review comment:
I don't think this is a good idea. Instead why not just create a
separate connectionOnServer and ensure that it contains just the mutations
related to SYSTEM.TASK?
##########
File path: phoenix-core/src/main/java/org/apache/phoenix/schema/task/Task.java
##########
@@ -141,12 +216,68 @@ public static void addTask(PhoenixConnection conn,
PTable.TaskType taskType, Str
PhoenixDatabaseMetaData.TASK_END_TS + ", " +
PhoenixDatabaseMetaData.TASK_DATA +
" ) VALUES(?,?,?,?,?,?,?,?,?)");
- stmt = setValuesToAddTaskPS(stmt, taskType, tenantId, schemaName,
tableName, taskStatus, data, priority, startTs, endTs);
- LOGGER.info("Adding task " + taskType + "," +tableName + "," +
taskStatus + "," + startTs, ","+endTs);
+ stmt = setValuesToAddTaskPS(stmt, systemTaskParams.getTaskType(),
+ systemTaskParams.getTenantId(),
+ systemTaskParams.getSchemaName(),
+ systemTaskParams.getTableName(),
+ systemTaskParams.getTaskStatus(), systemTaskParams.getData(),
+ systemTaskParams.getPriority(), systemTaskParams.getStartTs(),
+ systemTaskParams.getEndTs());
+ LOGGER.info("Adding task type: {} , tableName: {} , taskStatus: {}"
+ + " , startTs: {} , endTs: {}", systemTaskParams.getTaskType(),
+ systemTaskParams.getTableName(),
+ systemTaskParams.getTaskStatus(),
systemTaskParams.getStartTs(),
+ systemTaskParams.getEndTs());
} catch (SQLException e) {
throw new IOException(e);
}
- mutateSystemTaskTable(conn, stmt, accessCheckEnabled);
+ // if query is getting executed by client, do not execute and commit
+ // mutations
+ if (isCommitAllowed) {
Review comment:
Can we rename this variable? Sounds like it is set to `false` when we
don't really want to commit, but rather just get the stmt object which we later
use to get the `List<Mutation>`. This is used just for the index rebuild task
that gets triggered from the client-side, right? For the tasks that are
triggered from the server (i.e. from `MetaDataEndpointImpl` and `TaskRO`, we
can just directly call `Task.addTask()` right?
A more appropriate name would be `shouldCommit` or something.
##########
File path: phoenix-core/src/main/java/org/apache/phoenix/schema/task/Task.java
##########
@@ -75,6 +87,55 @@ public Void run() throws Exception {
}
}
+ private static List<Mutation> getMutationsForSystemTaskTable(
+ PhoenixConnection conn, PreparedStatement stmt,
+ boolean accessCheckEnabled) throws IOException {
+ // we need to mutate SYSTEM.TASK with HBase/login user if access is
enabled.
+ if (accessCheckEnabled) {
+ return User.runAsLoginUser(() -> {
+ final RpcCall rpcContext = RpcUtil.getRpcContext();
+ // setting RPC context as null so that user can be reset
+ try {
+ RpcUtil.setRpcContext(null);
+ stmt.execute();
+ // retrieve mutations for SYSTEM.TASK upsert query
+ Iterator<Pair<byte[], List<Mutation>>> iterator =
+ conn.getMutationState().toMutations();
+ List<Mutation> taskMutations = iterator.next().getSecond();
+ // we are expecting conn to be used for single upsert
+ // query on SYSTEM.TASK
+ if (iterator.hasNext()) {
+ throw new IOException("Provided connection should only
"
+ + "contain mutations related to SYSTEM.TASK
table");
+ }
+ return taskMutations;
+ } catch (SQLException e) {
+ throw new IOException(e);
+ } finally {
+ // setting RPC context back to original context of the RPC
+ RpcUtil.setRpcContext(rpcContext);
+ }
+ });
+ } else {
+ try {
+ stmt.execute();
Review comment:
Can we extract the this part into a small helper method? The `else` case
is similar to the `if` just that we aren't running with impersonation
##########
File path: phoenix-core/src/main/java/org/apache/phoenix/schema/task/Task.java
##########
@@ -75,6 +87,55 @@ public Void run() throws Exception {
}
}
+ private static List<Mutation> getMutationsForSystemTaskTable(
+ PhoenixConnection conn, PreparedStatement stmt,
+ boolean accessCheckEnabled) throws IOException {
+ // we need to mutate SYSTEM.TASK with HBase/login user if access is
enabled.
+ if (accessCheckEnabled) {
+ return User.runAsLoginUser(() -> {
+ final RpcCall rpcContext = RpcUtil.getRpcContext();
+ // setting RPC context as null so that user can be reset
+ try {
+ RpcUtil.setRpcContext(null);
+ stmt.execute();
+ // retrieve mutations for SYSTEM.TASK upsert query
+ Iterator<Pair<byte[], List<Mutation>>> iterator =
+ conn.getMutationState().toMutations();
+ List<Mutation> taskMutations = iterator.next().getSecond();
+ // we are expecting conn to be used for single upsert
+ // query on SYSTEM.TASK
+ if (iterator.hasNext()) {
Review comment:
Can it not be that the first `next()` gave you mutations corresponding
to some other upsert/delete? Let's just use our own connection
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Prevent doing direct upserts into SYSTEM.TASK from the client
> -------------------------------------------------------------
>
> Key: PHOENIX-6155
> URL: https://issues.apache.org/jira/browse/PHOENIX-6155
> Project: Phoenix
> Issue Type: Improvement
> Affects Versions: 5.0.0, 4.15.0
> Reporter: Chinmay Kulkarni
> Assignee: Viraj Jasani
> Priority: Major
> Fix For: 5.1.0, 4.16.0
>
> Attachments: PHOENIX-6155.4.x.000.patch, PHOENIX-6155.4.x.001.patch
>
>
> In environments with namespace-mapping enabled, we will have to grant write
> access to clients in order to make direct upserts into SYSTEM.TASK. Currently
> we add a task from the client-side
> [here|https://github.com/apache/phoenix/blob/4.x/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java#L4654].
> In order to implement other Jiras like
> [PHOENIX-6154|https://issues.apache.org/jira/browse/PHOENIX-6154] we also may
> need to interact with the SYSTEM.TASK table from the client-side.
> Instead of doing direct upserts into this table, we should add an endpoint on
> SYSTEM.TASK and clients should interact with that.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
