DTStack / DTStack/chunjun

[Question][restore] recovery from checkpoint failed

Open
#1,457 1 comment 0 reactions 0 assignees View on GitHub
question
Dominant language
Java
Stars
4.1k
Forks
1.7k
PR merge metrics
No merged PRs in 30d

Description

### Search before asking

- [X] I had searched in the [issues](https://github.com/DTStack/chunjun/issues) and found no similar question.

- [X] I had googled my question but i didn't get any help.

- [X] I had read the documentation: [ChunJun doc](https://dtstack.github.io/chunjun) but it didn't help me.

### Description

I try to restore the flink task from checkpoint, but the where condition is always 1=1, unable to get the id in checkpoint.

sample case
```bash
{
"job": {
"content": [
{
"reader": {
"table": {
"tableName": "dwd_dim_store_item"
},
"name": "mysqlreader",
"parameter": {
"column": ["*"],
"username": "root",
"password": "123456",
"connection": [
{
"jdbcUrl": [
"jdbc:mysql://localhost:12360/flink?useSSL=false"
],
"table": [
"dwd_dim_store_item"
]
}
]
}
},
"writer": {
"table": {
"tableName": "dwd_dim_store_item_1"
},
"name": "mysqlwriter",
"parameter": {
"username": "root",
"password": "123456",
"connection": [
{
"jdbcUrl": "jdbc:mysql://localhost:12360/flink?useSSL=false",
"table": [
"dwd_dim_store_item_1"
]
}
],
"writeMode": "insert",
"column": ["*"]
}
}
}
],
"setting": {
"restore": {
"isRestore": true,
"restoreColumnName": "id",
"restoreColumnIndex": 0
},
"speed": {
"channel": 1,
"bytes": 0
}
}
}
}
```

LocalTest.java
```java
argsList.add("-flinkConfDir");
argsList.add("/Users/kino/app/flink/flink-1.12.7/conf/");
argsList.add("-confProp");
// 从 savepoint 恢复
StringBuilder stringBuilder = new StringBuilder();
stringBuilder
.append("{")
.append("\"state.backend\":\"filesystem\"")
.append(", ")
.append("\"state.checkpoints.dir\":\"file:///Users/kino/app/flink/flink-1.12.7/ck\"")
.append(", ")
.append("\"state.checkpoints.num-retained\":\"10\"")
.append(", ")
.append("\"state.savepoints.dir\":\"file:///Users/kino/app/flink/flink-1.12.7/ck/f43f182f3a812edeab6363d2ad32f888/chk-5\"")
.append(", ")
.append("\"execution.savepoint.path\":\"file:///Users/kino/app/flink/flink-1.12.7/sp/savepoint-44fa76-88842a51f017\"")
.append(", ")
.append("}");
```

log
```text
2023-01-03 17:29:48,187 INFO DtInputFormatSourceFunction - Start initialize input format state, is restored:false
2023-01-03 17:29:48,205 INFO DtInputFormatSourceFunction - End initialize input format state
2023-01-03 17:29:52,036 INFO CheckpointCoordinator - Triggering checkpoint 2 (type=CHECKPOINT) @ 1672738192031 for job bdf8dda40064ffdc2a5f286f1edc93dc.
2023-01-03 17:29:52,105 INFO TaskSlotTableImpl - Free slot TaskSlot(index:0, state:RELEASING, resource profile: ResourceProfile{taskHeapMemory=51.200gb (54975581388 bytes), taskOffHeapMemory=51.200gb (54975581388 bytes), managedMemory=25.600mb (26843545 bytes), networkMemory=3.200mb (3355443 bytes)}, allocationId: 75e4a4900b930a8ef83542fdacbe4974, jobId: bdf8dda40064ffdc2a5f286f1edc93dc).
2023-01-03 17:29:52,127 INFO JdbcOutputFormat - format state:null
2023-01-03 17:29:52,128 INFO DtOutputFormatSinkFunction - OutputFormat format state:FormatState{numOfSubTask=0, state=null, metric={numWrite=LongCounter 0, conversionErrors=LongCounter 0, writeDuration=LongCounter 0, duplicateErrors=LongCounter 0, count=LongCounter 0, collectFailedCount=LongCounter 0, byteWrite=LongCounter 0, snapshotWrite=LongCounter 0, nullErrors=LongCounter 0, nErrors=LongCounter 0, otherErrors=LongCounter 0}, numberRead=0, numberWrite=0, jobId='null', fileIndex=-1}
2023-01-03 17:29:54,517 INFO DtInputFormatSourceFunction - InputFormat format state:FormatState{numOfSubTask=0, state=null, metric={byteRead=LongCounter 0, numRead=LongCounter 0, count=LongCounter 0, collectFailedCount=LongCounter 0, readDuration=LongCounter 0}, numberRead=0, numberWrite=0, jobId='null', fileIndex=-1}
2023-01-03 17:29:54,684 INFO CheckpointCoordinator - Completed checkpoint 2 for job bdf8dda40064ffdc2a5f286f1edc93dc (6001 bytes in 2648 ms).
2023-01-03 17:29:55,672 INFO JdbcInputFormat - Executing sql is: 'SELECT `id`, `tenant_id`, `store_id`, `shop_id`, `item_id`, `item_code`, `item_name`, `item_class`, `category_id_level1`, `category_id_level2`, `category_id_level3`, `normal_price`, `price`, `logistics`, `flag`, `pool_goods_type`, `promotion_type`, `out_date`, `good_date`, `stock_type`, `sales_flag`, `dc_shop_id`, `modify_time`, `etl_time` FROM `dwd_dim_store_item` WHERE 1=1 '
2023-01-03 17:29:57,013 INFO JdbcInputFormat - [JdbcInputFormat] open successfully,
inputSplit = JdbcInputSplit{mod=0, endLocation='null', startLocation='null', startLocationOfSplit='null', endLocationOfSplit='null', isPolling=false, splitStrategy='mod', rangeEndLocationOperator=' < '}GenericSplit (0/1),
[JdbcConf]:
{
"semantic" : "at-least-once",
"errorRecord" : 0,
"checkFormat" : true,
"parallelism" : 1,
"executeDdlAble" : false,
"pollingInterval" : 5000,
"increment" : false,
"orderBy" : true,
"flushIntervalMills" : 10000,
"polling" : false,
"querySql" : "SELECT `id`, `tenant_id`, `store_id`, `shop_id`, `item_id`, `item_code`, `item_name`, `item_class`, `category_id_level1`, `category_id_level2`, `category_id_level3`, `normal_price`, `price`, `logistics`, `flag`, `pool_goods_type`, `promotion_type`, `out_date`, `good_date`, `stock_type`, `sales_flag`, `dc_shop_id`, `modify_time`, `etl_time` FROM `dwd_dim_store_item` WHERE 1=1 ",
"mode" : "INSERT",
"password" : "******",
"metricPluginRoot" : "null/metrics",
"restoreColumn" : "id",
"restoreColumnIndex" : 0,
"connection" : [ {
"table" : [ "dwd_dim_store_item" ],
"jdbcUrl" : [ "jdbc:mysql://localhost:12360/flink?useSSL=false" ]
} ],
"table" : "dwd_dim_store_item",
"queryTimeOut" : 300,
"restoreColumnType" : "INT",
"fetchSize" : -2147483648,
"useMaxFunc" : false,
"pollingFromMax" : false,
"column" : [ {
"name" : "id",
"type" : "INT",
"index" : 0,
"notNull" : false,
"part" : false
}, {
"name" : "tenant_id",
"type" : "BIGINT",
"index" : 1,
"notNull" : false,
"part" : false
}, {
"name" : "store_id",
"type" : "VARCHAR",
"index" : 2,
"notNull" : false,
"part" : false
}, {
"name" : "shop_id",
"type" : "VARCHAR",
"index" : 3,
"notNull" : false,
"part" : false
}, {
"name" : "item_id",
"type" : "VARCHAR",
"index" : 4,
"notNull" : false,
"part" : false
}, {
"name" : "item_code",
"type" : "VARCHAR",
"index" : 5,
"notNull" : false,
"part" : false
}, {
"name" : "item_name",
"type" : "VARCHAR",
"index" : 6,
"notNull" : false,
"part" : false
}, {
"name" : "item_class",
"type" : "VARCHAR",
"index" : 7,
"notNull" : false,
"part" : false
}, {
"name" : "category_id_level1",
"type" : "BIGINT",
"index" : 8,
"notNull" : false,
"part" : false
}, {
"name" : "category_id_level2",
"type" : "BIGINT",
"index" : 9,
"notNull" : false,
"part" : false
}, {
"name" : "category_id_level3",
"type" : "BIGINT",
"index" : 10,
"notNull" : false,
"part" : false
}, {
"name" : "normal_price",
"type" : "DOUBLE",
"index" : 11,
"notNull" : false,
"part" : false
}, {
"name" : "price",
"type" : "DOUBLE",
"index" : 12,
"notNull" : false,
"part" : false
}, {
"name" : "logistics",
"type" : "DOUBLE",
"index" : 13,
"notNull" : false,
"part" : false
}, {
"name" : "flag",
"type" : "DOUBLE",
"index" : 14,
"notNull" : false,
"part" : false
}, {
"name" : "pool_goods_type",
"type" : "INT",
"index" : 15,
"notNull" : false,
"part" : false
}, {
"name" : "promotion_type",
"type" : "VARCHAR",
"index" : 16,
"notNull" : false,
"part" : false
}, {
"name" : "out_date",
"type" : "VARCHAR",
"index" : 17,
"notNull" : false,
"part" : false
}, {
"name" : "good_date",
"type" : "VARCHAR",
"index" : 18,
"notNull" : false,
"part" : false
}, {
"name" : "stock_type",
"type" : "VARCHAR",
"index" : 19,
"notNull" : false,
"part" : false
}, {
"name" : "sales_flag",
"type" : "VARCHAR",
"index" : 20,
"notNull" : false,
"part" : false
}, {
"name" : "dc_shop_id",
"type" : "VARCHAR",
"index" : 21,
"notNull" : false,
"part" : false
}, {
"name" : "modify_time",
"type" : "DATETIME",
"index" : 22,
"notNull" : false,
"part" : false
}, {
"name" : "etl_time",
"type" : "DATETIME",
"index" : 23,
"notNull" : false,
"part" : false
} ],
"errorPercentage" : -1,
"fieldNameList" : [ ],
"withNoLock" : false,
"increColumnIndex" : -1,
"allReplace" : false,
"splitStrategy" : "range",
"initReporter" : true,
"jdbcUrl" : "jdbc:mysql://localhost:12360/flink?useSSL=false",
"connectTimeOut" : 600,
"batchSize" : 1,
"speedBytes" : 0,
"rowSizeCalculatorType" : "objectSizeCalculator",
"metricPluginName" : "prometheus",
"properties" : {
"password" : "******",
"user" : "root",
"useCursorFetch" : "true",
"rewriteBatchedStatements" : "true"
},
"username" : "root"
}
```

ask:
Recovery from checkpoint fails in standalone mode,but in local mode, it can succeed

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.