[FLINK-36188] HBase disable buffer flush lose efficacy

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: hbase-3.0.0
• Fix Version/s: hbase-3.0.1, hbase-4.0.0
• Component/s: Connectors / HBase
• Labels: pull-request-available

Environment

Flink 1.16.3

• Assignee: MOBIN
• Reporter: MOBIN
• Created: 30/Aug/24 10:10
• Updated: 25/Sep/24 11:30
• Resolved: 25/Sep/24 10:23

Description

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

@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);
    } 

test failed

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] 

Issue Links

• links to: GitHub Pull Request #49