[jira] [Commented] (FLINK-36188) HBase disable buffer flush lose efficacy

["Xin Chen (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-36188?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17955079#comment-17955079 ] 

Xin Chen commented on FLINK-36188:
----------------------------------

I have a question here. If I have enabled the buffer flush, such as setting 'sink.buffer-flush.max-rows'='1000', but when I specify a table that definitely does not exist, that is, the table name field is a completely non-existent table, why does the flink task not report an error? It shows successful execution, and there is no error log in the task. Only when closed, there is a stack print.
{code:java}
tbEnv.executeSql("CREATE TABLE IF NOT EXISTS hTable (\n"
        + "rowkey STRING,\n"
        + "family1 ROW<q1 STRING,q2 STRING>,\n"
        + "family2 ROW<q3 STRING,q4 STRING>,\n"
        + "PRIMARY KEY (rowkey) NOT ENFORCED\n"
        + ") WITH (\n"
        + "'connector' = 'hbase-1.4',\n"
        + "'table-name' = 'mytable3',\n"  // mytable3 do not exist in Hbase Server!!!
        + "'sink.buffer-flush.max-rows' = '2',\n"
        + "'zookeeper.quorum' = 'hdp-zookeeper-hdp-zookeeper-0.hdp-zookeeper-hdp-zookeeper.cx-test.svc.cluster.local:2181'\n"
        + "    )"); {code}
 

> HBase disable buffer flush lose efficacy
> ----------------------------------------
>
>                 Key: FLINK-36188
>                 URL: https://issues.apache.org/jira/browse/FLINK-36188
>             Project: Flink
>          Issue Type: Bug
>          Components: Connectors / HBase
>    Affects Versions: hbase-3.0.0
>         Environment: Flink 1.16.3
>            Reporter: MOBIN
>            Assignee: MOBIN
>            Priority: Critical
>              Labels: pull-request-available
>             Fix For: hbase-3.0.1, hbase-4.0.0
>
>
> HBase table
> ||rowkey||col||
> |1|1|
> The user lookup joins the hbase table, adds 1 to the col value, and writes it back to hbase
> {code:java}
> @Test
> void testTableSinkDisabledBufferFlush() throws Exception {
>         StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
>         StreamTableEnvironment tEnv = StreamTableEnvironment.create(execEnv, streamSettings);        
>             tEnv.executeSql(
>                 "CREATE TABLE hTableForSink ("
>                         + " rowkey INT PRIMARY KEY NOT ENFORCED,"
>                         + " family1 ROW<col1 INT>"
>                         + ") WITH ("
>                         + " 'connector' = 'hbase-2.2',"
>                         + " 'sink.buffer-flush.max-size' = '0',"
>                         + " 'sink.buffer-flush.max-rows' = '0',"
>                         + " 'table-name' = '"
>                         + TEST_TABLE_6
>                         + "',"
>                         + " 'zookeeper.quorum' = '"
>                         + getZookeeperQuorum()
>                         + "'"
>                         + ")");       
>             String insert = "INSERT INTO hTableForSink VALUES(1, ROW(1))";
>         tEnv.executeSql(insert).await();        
>             tEnv.executeSql(
>                 "CREATE VIEW user_click AS "
>                         + " SELECT user_id, proctime() AS proc_time"
>                         + " FROM ( "
>                         + " VALUES(1), (1), (1), (1), (1)"
>                         + " ) AS t (user_id);");    
>     
>            tEnv.executeSql(
>                 "INSERT INTO hTableForSink SELECT "
>                         + "    user_id as rowkey,"
>                         + "    ROW(CAST(family1.col1 + 1 AS INT))"
>                         + " FROM user_click INNER JOIN hTableForSink"
>                         + " FOR SYSTEM_TIME AS OF user_click.proc_time"
>                         + " ON hTableForSink.rowkey = user_click.user_id;");        
>             tEnv.executeSql(
>                 "CREATE TABLE hTableForQuery ("
>                         + " rowkey INT PRIMARY KEY NOT ENFORCED,"
>                         + " family1 ROW<col1 INT>"
>                         + ") WITH ("
>                         + " 'connector' = 'hbase-2.2',"
>                         + " 'table-name' = '"
>                         + TEST_TABLE_6
>                         + "',"
>                         + " 'zookeeper.quorum' = '"
>                         + getZookeeperQuorum()
>                         + "'"
>                         + ")");
>         String query = "SELECT rowkey, family1.col1 FROM hTableForQuery";        
>             TableResult firstResult = tEnv.executeSql(query);
>         List<Row> firstResults = CollectionUtil.iteratorToList(firstResult.collect());
>         String firstExpected = "+I[1, 6]";
>         TestBaseUtils.compareResultAsText(firstResults, firstExpected);
>     } {code}
> test failed
> {code:java}
> org.junit.ComparisonFailure: Different elements in arrays: expected 1 elements and received 1
>  expected: [+I[1, 6]]
>  received: [+I[1, 2]] expected:<+I[1, [6]]> but was:<+I[1, [2]]>
> Expected :+I[1, 6]
> Actual   :+I[1, 2] {code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)