github-forks
/
flink-cdc-connectors
mirror of https://github.com/ververica/flink-cdc-connectors.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[mysql] Improve the code format in SignalEventDispatcher

Browse Source
pull/352/head
ili zh 4 years ago committed by GitHub
parent e32fb1e7fc
commit 73d110e4c6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 6 additions and 6 deletions
Show Stats Download Patch File Download Diff File

12
flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/debezium/dispatcher/SignalEventDispatcher.java
Unescape Escape View File

@ -37,7 +37,7 @@ import org.apache.kafka.connect.source.SourceRecord;
*/ */
public class SignalEventDispatcher { public class SignalEventDispatcher {
private static final SchemaNameAdjuster schemaNameAdjuster = SchemaNameAdjuster.create(); private static final SchemaNameAdjuster SCHEMA_NAME_ADJUSTER = SchemaNameAdjuster.create();
public static final String DATABASE_NAME = "db"; public static final String DATABASE_NAME = "db";
public static final String TABLE_NAME = "table"; public static final String TABLE_NAME = "table";
@ -66,13 +66,13 @@ public class SignalEventDispatcher {
this.queue = queue; this.queue = queue;
this.signalEventKeySchema = this.signalEventKeySchema =
SchemaBuilder.struct() SchemaBuilder.struct()
.name(schemaNameAdjuster.adjust(SIGNAL_EVENT_KEY_SCHEMA_NAME)) .name(SCHEMA_NAME_ADJUSTER.adjust(SIGNAL_EVENT_KEY_SCHEMA_NAME))
.field(SPLIT_ID_KEY, Schema.STRING_SCHEMA) .field(SPLIT_ID_KEY, Schema.STRING_SCHEMA)
.field(WATERMARK_SIGNAL, Schema.BOOLEAN_SCHEMA) .field(WATERMARK_SIGNAL, Schema.BOOLEAN_SCHEMA)
.build(); .build();
this.signalEventValueSchema = this.signalEventValueSchema =
SchemaBuilder.struct() SchemaBuilder.struct()
.name(schemaNameAdjuster.adjust(SIGNAL_EVENT_VALUE_SCHEMA_NAME)) .name(SCHEMA_NAME_ADJUSTER.adjust(SIGNAL_EVENT_VALUE_SCHEMA_NAME))
.field(SPLIT_ID_KEY, Schema.STRING_SCHEMA) .field(SPLIT_ID_KEY, Schema.STRING_SCHEMA)
.field(WATERMARK_KIND, Schema.STRING_SCHEMA) .field(WATERMARK_KIND, Schema.STRING_SCHEMA)
.field(BINLOG_FILENAME_OFFSET_KEY, Schema.STRING_SCHEMA) .field(BINLOG_FILENAME_OFFSET_KEY, Schema.STRING_SCHEMA)
@ -87,7 +87,7 @@ public class SignalEventDispatcher {
SourceRecord sourceRecord = SourceRecord sourceRecord =
new SourceRecord( new SourceRecord(
offsetContext.getPartition(), offsetContext.getPartition(),
offsetContext.getPartition(), offsetContext.getOffset(),
topic, topic,
signalEventKeySchema, signalEventKeySchema,
signalRecordKey(mySqlSplit.splitId()), signalRecordKey(mySqlSplit.splitId()),
@ -120,9 +120,9 @@ public class SignalEventDispatcher {
BINLOG_END; BINLOG_END;
public WatermarkKind fromString(String kindString) { public WatermarkKind fromString(String kindString) {
if ("LOW".equalsIgnoreCase(kindString)) { if (LOW.name().equalsIgnoreCase(kindString)) {
return LOW; return LOW;
} else if ("HIGH".equalsIgnoreCase(kindString)) { } else if (HIGH.name().equalsIgnoreCase(kindString)) {
return HIGH; return HIGH;
} else { } else {
return BINLOG_END; return BINLOG_END;

Write Preview
Loading…
Cancel
Save