You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2021/03/28 04:08:02 UTC
[flink] branch master updated: [FLINK-21702][sql-client] Support
option `sql-client.verbose` to print the exception stack
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e19d383 [FLINK-21702][sql-client] Support option `sql-client.verbose` to print the exception stack
e19d383 is described below
commit e19d383488a753483ac0f94ffda42592bc6f7765
Author: zhushang <17...@163.com>
AuthorDate: Fri Mar 26 14:54:51 2021 +0800
[FLINK-21702][sql-client] Support option `sql-client.verbose` to print the exception stack
This closes #15383
---
.../generated/sql_client_configuration.html | 6 ++
.../apache/flink/table/client/cli/CliClient.java | 6 +-
.../apache/flink/table/client/cli/CliStrings.java | 11 +++-
.../table/client/config/SqlClientOptions.java | 8 +++
.../flink/table/client/gateway/Executor.java | 20 +++++-
.../table/client/gateway/local/LocalExecutor.java | 7 +-
.../apache/flink/table/client/SqlClientTest.java | 75 ++++++++++++++++++----
.../flink/table/client/cli/CliClientTest.java | 14 +++-
.../flink/table/client/cli/CliResultViewTest.java | 8 ++-
.../flink/table/client/cli/TestingExecutor.java | 8 ++-
10 files changed, 139 insertions(+), 24 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/sql_client_configuration.html b/docs/layouts/shortcodes/generated/sql_client_configuration.html
index d9267b2..faf2a6a 100644
--- a/docs/layouts/shortcodes/generated/sql_client_configuration.html
+++ b/docs/layouts/shortcodes/generated/sql_client_configuration.html
@@ -20,5 +20,11 @@
<td><p>Enum</p>Possible values: [TABLE, CHANGELOG, TABLEAU]</td>
<td>Determine the mode when display the query result. The available values are ['table', 'tableau', 'changelog']. The 'table' mode materializes results in memory and visualizes them in a regular, paginated table representation. The 'changelog' mode does not materialize results and visualizes the result stream that is produced by a continuous query. The 'tableau' mode is more like a traditional way which will display the results in the screen directly with a tableau format. </td>
</tr>
+ <tr>
+ <td><h5>sql-client.verbose</h5><br> <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Determine whether to output the verbose output to the console. If set the option true, it will print the exception stack. Otherwise, it only output the cause.</td>
+ </tr>
</tbody>
</table>
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
index f2e037e..b512bdc 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.client.SqlClient;
import org.apache.flink.table.client.SqlClientException;
+import org.apache.flink.table.client.config.SqlClientOptions;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
@@ -354,7 +355,7 @@ public class CliClient implements AutoCloseable {
}
// show all properties
else {
- final Map<String, String> properties = executor.getSessionProperties(sessionId);
+ final Map<String, String> properties = executor.getSessionConfigMap(sessionId);
if (properties.isEmpty()) {
terminal.writer()
.println(CliStrings.messageInfo(CliStrings.MESSAGE_EMPTY).toAnsi());
@@ -462,7 +463,8 @@ public class CliClient implements AutoCloseable {
private void printExecutionException(Throwable t) {
final String errorMessage = CliStrings.MESSAGE_SQL_EXECUTION_ERROR;
LOG.warn(errorMessage, t);
- terminal.writer().println(CliStrings.messageError(errorMessage, t).toAnsi());
+ boolean isVerbose = executor.getSessionConfig(sessionId).get(SqlClientOptions.VERBOSE);
+ terminal.writer().println(CliStrings.messageError(errorMessage, t, isVerbose).toAnsi());
terminal.flush();
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
index e750438..782a4ed 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
@@ -18,6 +18,8 @@
package org.apache.flink.table.client.cli;
+import org.apache.flink.util.ExceptionUtils;
+
import org.jline.utils.AttributedString;
import org.jline.utils.AttributedStringBuilder;
import org.jline.utils.AttributedStyle;
@@ -297,14 +299,17 @@ public final class CliStrings {
.toAttributedString();
}
- public static AttributedString messageError(String message, Throwable t) {
+ public static AttributedString messageError(String message, Throwable t, boolean isVerbose) {
while (t.getCause() != null
&& t.getCause().getMessage() != null
&& !t.getCause().getMessage().isEmpty()) {
t = t.getCause();
}
- return messageError(message, t.getClass().getName() + ": " + t.getMessage());
- // return messageError(message, ExceptionUtils.stringifyException(t));
+ if (isVerbose) {
+ return messageError(message, ExceptionUtils.stringifyException(t));
+ } else {
+ return messageError(message, t.getClass().getName() + ": " + t.getMessage());
+ }
}
public static AttributedString messageError(String message) {
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java
index fc684ed..bafe2d3 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java
@@ -45,4 +45,12 @@ public class SqlClientOptions {
+ "The 'table' mode materializes results in memory and visualizes them in a regular, paginated table representation. "
+ "The 'changelog' mode does not materialize results and visualizes the result stream that is produced by a continuous query. "
+ "The 'tableau' mode is more like a traditional way which will display the results in the screen directly with a tableau format. ");
+
+ @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
+ public static final ConfigOption<Boolean> VERBOSE =
+ ConfigOptions.key("sql-client.verbose")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Determine whether to output the verbose output to the console. If set the option true, it will print the exception stack. Otherwise, it only output the cause.");
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
index f4fc20e..66aa637 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.client.gateway;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
@@ -51,8 +52,23 @@ public interface Executor {
*/
void closeSession(String sessionId) throws SqlExecutionException;
- /** Lists all session properties that are defined by the executor and the session. */
- Map<String, String> getSessionProperties(String sessionId) throws SqlExecutionException;
+ /**
+ * Returns a copy of {@link Map} of all session configurations that are defined by the executor
+ * and the session.
+ *
+ * <p>Both this method and {@link #getSessionConfig(String)} return the same configuration set,
+ * but different return type.
+ */
+ Map<String, String> getSessionConfigMap(String sessionId) throws SqlExecutionException;
+
+ /**
+ * Returns a {@link ReadableConfig} of all session configurations that are defined by the
+ * executor and the session.
+ *
+ * <p>Both this method and {@link #getSessionConfigMap(String)} return the same configuration
+ * set, but different return type.
+ */
+ ReadableConfig getSessionConfig(String sessionId) throws SqlExecutionException;
/**
* Reset all the properties for the given session identifier.
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
index bb457ff..f3e770f 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
@@ -135,11 +135,16 @@ public class LocalExecutor implements Executor {
}
@Override
- public Map<String, String> getSessionProperties(String sessionId) throws SqlExecutionException {
+ public Map<String, String> getSessionConfigMap(String sessionId) throws SqlExecutionException {
return getSessionContext(sessionId).getConfigMap();
}
@Override
+ public ReadableConfig getSessionConfig(String sessionId) throws SqlExecutionException {
+ return getSessionContext(sessionId).getReadableConfig();
+ }
+
+ @Override
public void resetSessionProperties(String sessionId) throws SqlExecutionException {
SessionContext context = getSessionContext(sessionId);
context.reset();
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java
index 128e126..7e1d754 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java
@@ -43,6 +43,7 @@ import java.util.Objects;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.not;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
@@ -91,7 +92,7 @@ public class SqlClientTest {
}
@After
- public void after() throws InterruptedException {
+ public void after() {
System.setOut(originalPrintStream);
System.setIn(originalInputStream);
CommonTestUtils.setEnv(originalEnv);
@@ -101,43 +102,43 @@ public class SqlClientTest {
return testOutputStream.toString();
}
- @Test(timeout = 20000)
- public void testEmbeddedWithOptions() throws InterruptedException {
+ @Test
+ public void testEmbeddedWithOptions() {
String[] args = new String[] {"embedded", "-hist", historyPath};
SqlClient.main(args);
assertThat(getStdoutString(), containsString("Command history file path: " + historyPath));
}
- @Test(timeout = 20000)
- public void testEmbeddedWithLongOptions() throws InterruptedException {
+ @Test
+ public void testEmbeddedWithLongOptions() {
String[] args = new String[] {"embedded", "--history", historyPath};
SqlClient.main(args);
assertThat(getStdoutString(), containsString("Command history file path: " + historyPath));
}
- @Test(timeout = 20000)
- public void testEmbeddedWithoutOptions() throws InterruptedException {
+ @Test
+ public void testEmbeddedWithoutOptions() {
String[] args = new String[] {"embedded"};
SqlClient.main(args);
assertThat(getStdoutString(), containsString("Command history file path"));
}
- @Test(timeout = 20000)
- public void testEmptyOptions() throws InterruptedException {
+ @Test
+ public void testEmptyOptions() {
String[] args = new String[] {};
SqlClient.main(args);
assertThat(getStdoutString(), containsString("Command history file path"));
}
- @Test(timeout = 20000)
- public void testUnsupportedGatewayMode() throws Exception {
+ @Test
+ public void testUnsupportedGatewayMode() {
String[] args = new String[] {"gateway"};
thrown.expect(SqlClientException.class);
thrown.expectMessage("Gateway mode is not supported yet.");
SqlClient.main(args);
}
- @Test(timeout = 20000)
+ @Test
public void testPrintHelpForUnknownMode() throws IOException {
String[] args = new String[] {"unknown"};
SqlClient.main(args);
@@ -146,4 +147,54 @@ public class SqlClientTest {
final String help = FileUtils.readFileUtf8(new File(url.getFile()));
assertEquals(help, getStdoutString());
}
+
+ @Test
+ public void testErrorMessage() {
+ // prepare statements which will throw exception
+ String stmts =
+ "CREATE TABLE T (a int) WITH ('connector' = 'invalid');\n"
+ + "SELECT * FROM T;\n"
+ + "QUIT;\n";
+ System.setIn(new ByteArrayInputStream(stmts.getBytes(StandardCharsets.UTF_8)));
+ String[] args = new String[] {};
+ SqlClient.main(args);
+ String output = getStdoutString();
+ assertThat(
+ output,
+ containsString(
+ "org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'invalid'"));
+
+ // shouldn't contain error stack
+ String[] errorStack =
+ new String[] {
+ "at org.apache.flink.table.factories.FactoryUtil.discoverFactory",
+ "at org.apache.flink.table.factories.FactoryUtil.createTableSource"
+ };
+ for (String stack : errorStack) {
+ assertThat(output, not(containsString(stack)));
+ }
+ }
+
+ @Test
+ public void testVerboseErrorMessage() {
+ // prepare statements which will throw exception
+ String stmts =
+ "CREATE TABLE T (a int) WITH ('connector' = 'invalid');\n"
+ + "SET sql-client.verbose=true;\n"
+ + "SELECT * FROM T;\n"
+ + "QUIT;\n";
+ System.setIn(new ByteArrayInputStream(stmts.getBytes(StandardCharsets.UTF_8)));
+ String[] args = new String[] {};
+ SqlClient.main(args);
+ String output = getStdoutString();
+ String[] errors =
+ new String[] {
+ "org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'invalid'",
+ "at org.apache.flink.table.factories.FactoryUtil.discoverFactory",
+ "at org.apache.flink.table.factories.FactoryUtil.createTableSource"
+ };
+ for (String error : errors) {
+ assertThat(output, containsString(error));
+ }
+ }
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
index f9f32b8..fa65980 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.client.cli;
import org.apache.flink.client.cli.DefaultCLI;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.environment.TestingJobClient;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
@@ -214,12 +215,18 @@ public class CliClientTest extends TestLogger {
public void closeSession(String sessionId) throws SqlExecutionException {}
@Override
- public Map<String, String> getSessionProperties(String sessionId)
+ public Map<String, String> getSessionConfigMap(String sessionId)
throws SqlExecutionException {
return null;
}
@Override
+ public ReadableConfig getSessionConfig(String sessionId) throws SqlExecutionException {
+ SessionContext context = this.sessionMap.get(sessionId);
+ return context.getReadableConfig();
+ }
+
+ @Override
public void resetSessionProperties(String sessionId) throws SqlExecutionException {}
@Override
@@ -228,7 +235,10 @@ public class CliClientTest extends TestLogger {
@Override
public void setSessionProperty(String sessionId, String key, String value)
- throws SqlExecutionException {}
+ throws SqlExecutionException {
+ SessionContext context = this.sessionMap.get(sessionId);
+ context.set(key, value);
+ }
@Override
public TableResult executeOperation(String sessionId, Operation operation)
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
index a4069aa..c2fc27a 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.table.client.cli;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.catalog.Column;
@@ -140,12 +141,17 @@ public class CliResultViewTest {
}
@Override
- public Map<String, String> getSessionProperties(String sessionId)
+ public Map<String, String> getSessionConfigMap(String sessionId)
throws SqlExecutionException {
return null;
}
@Override
+ public ReadableConfig getSessionConfig(String sessionId) throws SqlExecutionException {
+ return null;
+ }
+
+ @Override
public void resetSessionProperties(String sessionId) throws SqlExecutionException {}
@Override
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
index 526ffe8..1850617 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
@@ -17,6 +17,7 @@
package org.apache.flink.table.client.cli;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.client.cli.utils.SqlParserHelper;
import org.apache.flink.table.client.gateway.Executor;
@@ -95,11 +96,16 @@ class TestingExecutor implements Executor {
public void closeSession(String sessionId) throws SqlExecutionException {}
@Override
- public Map<String, String> getSessionProperties(String sessionId) throws SqlExecutionException {
+ public Map<String, String> getSessionConfigMap(String sessionId) throws SqlExecutionException {
throw new UnsupportedOperationException("Not implemented.");
}
@Override
+ public ReadableConfig getSessionConfig(String sessionId) throws SqlExecutionException {
+ return null;
+ }
+
+ @Override
public void resetSessionProperties(String sessionId) throws SqlExecutionException {
throw new UnsupportedOperationException("Not implemented.");
}