You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "kwokhh (via GitHub)" <gi...@apache.org> on 2025/01/21 03:32:33 UTC
[I] [Feature] Flink cdc paimon ingestion recovery issue [paimon]
kwokhh opened a new issue, #4969:
URL: https://github.com/apache/paimon/issues/4969
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.
### Motivation
I face some issue when loading a big database from mysql to paimon just using one flink paimon cdc ingestion job. If flink ingestion job encounters an exception triggered by loading one of the tables in the database, then the whole flink job will stop and affect the ingestion work of the other table ingestion process. If I encounter the exception which requires me to restart the job without initial savepoint to solve it, is there any configuration to avoid initial loading all table in the job configured and just initial loading the table which trigger the exception?
### Solution
No Solution
### Anything else?
Using paimon 0.9 and flink1.20
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Flink cdc paimon ingestion recovery issue [paimon]
Posted by "yangjf2019 (via GitHub)" <gi...@apache.org>.
yangjf2019 commented on issue #4969:
URL: https://github.com/apache/paimon/issues/4969#issuecomment-2606136013
Hi @kwokhh ,Thanks for your issue!Would you like to provide more information for this problem? For example, flink code, logs,env and so on.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Flink cdc paimon ingestion recovery issue [paimon]
Posted by "yangjf2019 (via GitHub)" <gi...@apache.org>.
yangjf2019 commented on issue #4969:
URL: https://github.com/apache/paimon/issues/4969#issuecomment-2652492442
Thanks for your information! and from the information provided in the yaml, I see that the versions of flink and paimon are not the ones that are running with errors.
```
flinkVersion: v1_17
jarURI: local:////opt/flink/usrlib/paimon-flink-action-0.8.2.jar
```
<img width="782" alt="Image" src="https://github.com/user-attachments/assets/57d007ca-289e-4e2d-8a69-1aab94278da3" />
--
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.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Flink cdc paimon ingestion recovery issue [paimon]
Posted by "kwokhh (via GitHub)" <gi...@apache.org>.
kwokhh commented on issue #4969:
URL: https://github.com/apache/paimon/issues/4969#issuecomment-2650072070
Hi @yangjf2019 the flink deployment is described below:
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: <name>
spec:
image: <name>
mode: native
flinkVersion: v1_17
flinkConfiguration:
taskmanager.memory.jvm-metaspace.size: 512 mb
jobmanager.memory.jvm-metaspace.size: 512 mb
metrics.job.status.enable: "STATE"
metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
taskmanager.numberOfTaskSlots: "4"
taskmanager.memory.managed.fraction: "0.1"
high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
state.savepoints.dir: <dir>
state.checkpoints.dir: <dir>
high-availability.storageDir: <dir>
kubernetes.hadoop.conf.config-map.name: <name>
kubernetes.operator.periodic.savepoint.interval: 30 min
kubernetes.operator.savepoint.history.max.age: 14 d
kubernetes.operator.savepoint.history.max.count: "5"
kubernetes.operator.job.upgrade.last-state.max.allowed.checkpoint.age: 30 min
execution.checkpointing.unaligned.enabled: "false"
execution.checkpointing.tolerable-failed-checkpoints: "100"
execution.checkpointing.interval: 15 min
execution.checkpointing.min-pause: 2 min
execution.checkpointing.timeout: 10 min
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: "2147483647"
restart-strategy.fixed-delay.delay: 120 s
state.checkpoints.num-retained: "10"
pipeline.object-reuse: "true"
akka.ask.timeout: 10m
client.timeout: 15 min
e
akka.framesize: 41943040b
table.local-time-zone: <region>
podTemplate:
apiVersion: v1
kind: Pod
metadata:
name: pod-template
spec:
imagePullSecrets:
- name: <name>
containers:
- name: flink-main-container
ports:
- name: metrics
containerPort: 9249
protocol: TCP
serviceAccount: flink
jobManager:
replicas: 1
resource:
memory: "32G"
cpu: 4
taskManager:
replicas: 2
resource:
memory: "32G"
cpu: 4
restartNonce: 0
job:
jarURI: local:////opt/flink/usrlib/paimon-flink-action-0.8.2.jar
entryClass: "org.apache.paimon.flink.action.FlinkActions"
#
args: [
"mysql_sync_database",
"--warehouse", "<link>",
"--database" , "<db>",
"--mysql_conf", "hostname=<hostname>",
"--mysql_conf", "port=<port>",
"--mysql_conf", "username=<name>",
"--mysql_conf", "password=<password>",
"--mysql_conf", "database-name=<name>",
"--mysql_conf", "server-id=<id>",
"--mysql_conf", "server-time-zone=<zone>",
"--including_tables", "<tables>",
"--catalog_conf", "metastore=filesystem",
"--catalog_conf", "case-sensitive=false",
"--table_conf", "bucket=4",
"--table_conf", "changelog-producer=input"
]
# parallelism: 2
upgradeMode: last-state
allowNonRestoredState: true
state: running
savepointTriggerNonce: 1
ingress:
template: "<link>"
className: "nginx"
annotations:
nginx.ingress.kubernetes.io/rewrite-target: "/$2"
logConfiguration:
log4j-console.properties: |+
# This affects logging for both user code and Flink
rootLogger.level = INFO
rootLogger.appenderRef.console.ref = ConsoleAppender
rootLogger.appenderRef.rolling.ref = RollingFileAppender
logger.akka.name = <name>
logger.akka.level = INFO
logger.kafka.name= org.apache.kafka
logger.kafka.level = INFO
logger.hadoop.name = org.apache.hadoop
logger.hadoop.level = INFO
logger.zookeeper.name = org.apache.zookeeper
logger.zookeeper.level = INFO
appender.console.name = ConsoleAppender
appender.console.type = CONSOLE
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %replace{%m}{(password['|\"]*\\s*[=|:]\\s*['|\"]*)[^'\",\r\n\\s]+('|\"|,|\r\n|\r|\n|\\s|$)}{$1****$2}%n
# Log all infos in the given rolling file
appender.rolling.name = RollingFileAppender
appender.rolling.type = RollingFile
appender.rolling.append = false
appender.rolling.fileName = ${sys:log.file}
appender.rolling.filePattern = ${sys:log.file}.%i
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %replace{%m}{(password['|\"]*\\s*[=|:]\\s*['|\"]*)[^'\",\r\n\\s]+('|\"|,|\r\n|\r|\n|\\s|$)}{$1****$2}%n
appender.rolling.policies.type = Policies
appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
appender.rolling.policies.size.size=100MB
appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.max = 10
logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
logger.netty.level = OFF
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org