[jira] [Created] (FLINK-34809) [mysql] Add notifications to Slack when the Snapshot phase ends or Binlog stream phase begins

["Flink CDC Issue Import (Jira)" <ji...@apache.org>]

Flink CDC Issue Import created FLINK-34809:
----------------------------------------------

             Summary: [mysql] Add notifications to Slack when the Snapshot phase ends or Binlog stream phase begins
                 Key: FLINK-34809
                 URL: https://issues.apache.org/jira/browse/FLINK-34809
             Project: Flink
          Issue Type: Improvement
          Components: Flink CDC
            Reporter: Flink CDC Issue Import


### Search before asking

- [X] I searched in the [issues|https://github.com/ververica/flink-cdc-connectors/issues] and found nothing similar.


### Motivation

On our team, we use Flink CDC to perform MySql CDC.

Since [there is no Snapshot Only mode for MySql yet|https://github.com/ververica/flink-cdc-connectors/issues/1687], we had a need to be notified when a snapshot is completed and when the binlog stream is started.

To accomplish this, we **implemented a notification when a snapshot ends and when a binlog stream starts with  GTIDs.**

--- 

Here's the team's use case in more detail 
1. We set parallelism to 2 or more for large tables.
2. And we send change event log to kafka to use Debezium's JDBC Sink Connector, which supports [Schema Evolution|https://debezium.io/documentation/reference/stable/connectors/jdbc.html#jdbc-schema-evolution).
3. Sinking to Kafka is slower than MySqlSource operator, so we give more paralleisms to sink operator more parallelism than MySqlSource.
4. In this case, the transfer is done in rebalance mode from source operator to sink operator, so the order for the same PK is not guaranteed when transferring binlogs.
5. So we restart the job based on GTIDs with parallelism equal to 1 at the end of the snapshot phase .

To do this, we needed (1) to be notified that the snapshot ended and the binlog stream started, and (2) to know from which GTIDs the binlog stream started.

---

This is portion of our code. We **assumed that we only capture 1 table per flink cdc job.**
```scala
  def getMySQLSourceOperator(]: MySqlSource[String] = {
    MySqlSource.builder[String|)
      .hostname(mySqlConfig.host)
      .port(mySqlConfig.port)
      .serverTimeZone(mySqlConfig.timeZone)
      .databaseList(mySqlConfig.database)
      .tableList(mySqlConfig.table)
      .username(mySqlConfig.user)
      .serverId(mySqlConfig.serverIdRange)
      .password(mySqlConfig.password)
      .startupOptions(mySqlConfig.startupMode)
      .fetchSize(mySqlConfig.fetchSize)
      .splitSize(mySqlConfig.splitSize)
      .chunkKeyColumn(new ObjectPath(mySqlConfig.database, mySqlConfig.table), mySqlConfig.chunkKeyColumn)
      .connectionPoolSize(mySqlConfig.poolSize)
      .scanNewlyAddedTableEnabled(false)
      .includeSchemaChanges(false)
      .debeziumProperties(mySqlConfig.dbzProps)
      .closeIdleReaders(true)
      .notifySnapshotToBinlogSwitch("slack-hook-url") // here what we implemented
      .deserializer(new JsonDebeziumDeserializationSchema(true, mySqlConfig.jsonConverterProps))
      .build()
  }
```

--- 

We can't share a real picture of the notification, because our company recommends using in-house tools rather than Slack(🥲🥲🥲] and has some security policy.

But it looks something like the format below!
- Snapshot finished notifiaction.
```
[SNAPSHOT FINISHED]
 Database: test_database
 Table: test_table
```
- Binlog stream start notification.
```
[BINLOG STREAM START]
 Database: test_database
 Table: test_table
 GTIDs: 3bda59bb-2fc8-11eb-855f-fa163e2550e3:1-128377129,3c3a6a1b-c931-11ed-b0db-b4055dec129e:1-273703345,901d637c-8add-11eb-8e3f-b4055d3355a6:1-3641352422,b46b8251-5254-11ed-a648-d0946637df48:1-1069556331,db449f07-c53e-11e8-b8c2-d094663d3d1d:1-3543229125,e8d62f95-c77a-11e8-9270-d0946637df48:1-5715021533
```


### Solution

Our implementation and PR is here. https://github.com/ververica/flink-cdc-connectors/pull/2453

### Alternatives

_No response_

### Anything else?

_No response_

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!

---------------- Imported from GitHub ----------------
Url: https://github.com/apache/flink-cdc/issues/2454
Created by: [SML0127|https://github.com/SML0127]
Labels: enhancement, 
Created at: Sat Sep 02 14:34:50 CST 2023
State: open




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