Realtime Compute for Apache Flink:Data synchronization issues

JSON schema changes from Kafka to Hologres

The conventional solution is to stop the Flink deployment, modify the code and the Hologres table schema, and then redeploy. However, this process can cause data processing delays and introduce unnecessary errors.

Realtime Compute for Apache Flink provides the following optimizations to address this issue:

• Automatic schema evolution for Kafka JSON format: If the JSON schema changes, the system automatically synchronizes the new schema to Hologres without requiring you to stop the deployment or modify SQL statements.

• Type inference for Kafka JSON format: You do not need to declare the data types of JSON fields in the DDL.

• Recursive expansion of nested JSON fields: For example, for the JSON data {"nested": {"col": true}}, the col field is expanded to nested.col.

Error: Encountered change event for table xxx.xxx whose schema isn't known to this connector

• Problem

  202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader    [] - Encountered change event 'Event{header=EventHeaderV4{timestamp=xxx, eventType=TABLE_MAP, serverId=xxx, headerLength=xxx, dataLength=xxx, nextPosition=xxx, flags=xxx}, data=TableMapEventData{tableId=xxx, database='xxx', table='xxx',
  columnTypes=xxx, xxx..., columnMetadata=xxx,xxx,..., columnNullability={xxx,xxx,...}, eventMetadata=null}}' at offset {ts_sec=xxx, file=mysql-bin.xxx, pos=xxx,
  gtids=xxx, ...server_id=xxx, event=xxx} for table xxx.xxx whose schema isn't known to this connector. One possible cause is an incomplete database history topic. Take
  a new snapshot in this case.
  Use the mysqlbinlog tool to view the problematic event: mysqlbinlog --start-position=30946 --stop-position=31028 --verbose mysql-bin.004419
  202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader    [] - Error during binlog processing. Last offset stored = null, binlog
  reader near position = mysql-bin.xxx/xxx
  202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader    [] - Failed due to error: Error processing binlog event
  org.apache.kafka.connect.errors.ConnectException: Encountered change event for table statistic.apk info whose schema isn't known to this connector
  	at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241)
  	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:210)
  	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:607)
  	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at com.github.shyiko.mysql.binlog.BinaryLogClient.notifyEventListeners(BinaryLogClient.java:1104)
  	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:955)
  	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:595)
  	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at com.github.shyiko.mysql.binlog.BinaryLogClientIMT.run(BinaryLogClient.java:839)
  	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at java.lang.Thread.run(Thread.java:834) [?:1.8.0_102]
  Caused by: org.apache.kafka.connect.errors.ConnectException: Encountered change event for table xxx.xxx whose schema isn't known to this connector
  	at io.debezium.connector.mysql.BinlogReader.informAboutUnknownTableIfRequired(BinlogReader.java:875)
  	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at io.debezium.connector.mysql.BinlogReader.handleUpdateTableMetadata(BinlogReader.java:849)
  	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:590)
  	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
  	... 5 more

• Cause

  When you use the CDAS or CTAS feature to read new tables, this error can occur for the following reasons:

  • The database account may lack the required permissions on some databases used by the deployment.

  • The deployment is configured with 'debezium.snapshot.mode'='never'. This setting causes the connector to read from the beginning of the binlog. If the table schema associated with an early change event in the binlog does not match the current table schema, this error occurs.

  • This error may occur if Debezium encounters a change that it cannot parse, such as DEFAULT (now()).

• Solution

  • Ensure the database account has all the database permissions that the deployment requires. These permissions typically include reading, writing, and modifying data, and creating and modifying table schemas and tables. For more information, see the corresponding connector documentation.

  • Use the 'debezium.inconsistent.schema.handling.mode' = 'warn' parameter to prevent errors. Do not use 'debezium.snapshot.mode'='never'.

  • Continue to monitor the logs. Look for io.debezium.connector.mysql.MySqlSchema WARN messages, which provide details about the specific changes that cannot be parsed. For example, Debezium cannot parse DEFAULT (now()).

Error: Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax

• Details

  An error occurs when deploying or starting a deployment:

  USE CATALOG `test-holo`;
  CREATE TABLE IF NOT EXISTS test_sync
  AS TABLE `test-mysql`.`flink_test`.`test_sync`
  /*+ OPTIONS('server-id'='9021-9030') */;
  Problem: test-ctas
  org.apache.flink.table.api.TableException: Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax.
      at org.apache.flink.table.planner.plan.nodes.exec.processor.EvolvingTableSourceMergeProcessor$EvolvingNodeMergeVisitor.visit
  (EvolvingTableSourceMergeProcessor.java:285)
      at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.accept(ExecNodeBase.java:249)
      at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.lambda$visitInputs$0(ExecNodeVisitorImpl.java:34)
      at java.util.ArrayList.forEach(ArrayList.java:1249)
      at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.visitInputs(ExecNodeVisitorImpl.java:34)
      at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.visit(ExecNodeVisitorImpl.java:30)
      at org.apache.flink.table.planner.plan.nodes.exec.processor.EvolvingTableSourceMergeProcessor$EvolvingNodeMergeVisitor.visit
  (EvolvingTableSourceMergeProcessor.java:263)
      at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.accept(ExecNodeBase.java:249)
      at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.lambda$visitInputs$0(ExecNodeVisitorImpl.java:34)
      at java.util.ArrayList.forEach(ArrayList.java:1249)
      at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.visitInputs(ExecNodeVisitorImpl.java:34)

• Cause

  CTAS and CDAS deployments do not support MiniBatch operations. However, the table.exec.mini-batch.enabled: 'true' configuration is enabled in the SQL deployment.

• Solution

  Remove the MiniBatch-related configurations from the SQL deployment.

  • For failed deployments

    In the Other Configuration section on the Deployment Defaults tab of the Configurations page, delete the MiniBatch-related configuration information or set table.exec.mini-batch.enabled to false. After you save the changes, re-create and deploy the SQL deployment.

  • For existing deployments

    If a deployment skipped the in-depth pre-deployment check and is completed, go to the Deployment Details tab on the Deployments page. In the Other Configuration section of Parameters, delete the MiniBatch configuration information or set table.exec.mini-batch.enabled to false, and restart the deployment.