-
Notifications
You must be signed in to change notification settings - Fork 1.9k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[tidb] add tidb cdc connector #898
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks @eastfisher for the contribution, I left some comments. Please let me know If you need any help.
...ava/com/ververica/cdc/connectors/tidb/table/RowDataTiKVChangeEventDeserializationSchema.java
Outdated
Show resolved
Hide resolved
flink-connector-tidb-cdc/src/main/java/org/tikv/common/codec/RowV2.java
Outdated
Show resolved
Hide resolved
flink-connector-tidb-cdc/src/main/java/org/tikv/common/codec/TableCodecV1.java
Outdated
Show resolved
Hide resolved
flink-connector-tidb-cdc/src/main/java/org/tikv/common/codec/TableCodecV2.java
Outdated
Show resolved
Hide resolved
flink-connector-tidb-cdc/src/main/java/org/tikv/common/codec/TableCodecV2.java
Outdated
Show resolved
Hide resolved
# See the License for the specific language governing permissions and | ||
# limitations under the License. | ||
|
||
com.ververica.cdc.connectors.tidb.table.TiDBTableSourceFactory |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
please help add one more blank line
...ctor-tidb-cdc/src/test/java/com/ververica/cdc/connectors/tidb/table/TiDBConnectorITCase.java
Outdated
Show resolved
Hide resolved
public static final ConfigOption<String> TIKV_PD_ADDRESSES = | ||
ConfigOptions.key(ConfigUtils.TIKV_PD_ADDRESSES) | ||
.stringType() | ||
.noDefaultValue() | ||
.withDescription("TiKV cluster's PD address"); | ||
|
||
public static final ConfigOption<Long> TIKV_GRPC_TIMEOUT = | ||
ConfigOptions.key(ConfigUtils.TIKV_GRPC_TIMEOUT) | ||
.longType() | ||
.noDefaultValue() | ||
.withDescription("TiKV GRPC timeout in ms"); | ||
|
||
public static final ConfigOption<Long> TIKV_GRPC_SCAN_TIMEOUT = | ||
ConfigOptions.key(ConfigUtils.TIKV_GRPC_SCAN_TIMEOUT) | ||
.longType() | ||
.noDefaultValue() | ||
.withDescription("TiKV GRPC scan timeout in ms"); | ||
|
||
public static final ConfigOption<Integer> TIKV_BATCH_GET_CONCURRENCY = | ||
ConfigOptions.key(ConfigUtils.TIKV_BATCH_GET_CONCURRENCY) | ||
.intType() | ||
.noDefaultValue() | ||
.withDescription("TiKV GRPC batch get concurrency"); | ||
|
||
public static final ConfigOption<Integer> TIKV_BATCH_PUT_CONCURRENCY = | ||
ConfigOptions.key(ConfigUtils.TIKV_BATCH_PUT_CONCURRENCY) | ||
.intType() | ||
.noDefaultValue() | ||
.withDescription("TiKV GRPC batch put concurrency"); | ||
|
||
public static final ConfigOption<Integer> TIKV_BATCH_SCAN_CONCURRENCY = | ||
ConfigOptions.key(ConfigUtils.TIKV_BATCH_SCAN_CONCURRENCY) | ||
.intType() | ||
.noDefaultValue() | ||
.withDescription("TiKV GRPC batch scan concurrency"); | ||
|
||
public static final ConfigOption<Integer> TIKV_BATCH_DELETE_CONCURRENCY = | ||
ConfigOptions.key(ConfigUtils.TIKV_BATCH_DELETE_CONCURRENCY) | ||
.intType() | ||
.noDefaultValue() | ||
.withDescription("TiKV GRPC batch delete concurrency"); | ||
|
||
public static TiConfiguration getTiConfiguration(final Map<String, String> options) { | ||
final Configuration configuration = Configuration.fromMap(options); | ||
|
||
final TiConfiguration tiConf = | ||
configuration | ||
.getOptional(TIKV_PD_ADDRESSES) | ||
.map(TiConfiguration::createDefault) | ||
.orElseGet(TiConfiguration::createDefault); | ||
|
||
configuration.getOptional(TIKV_GRPC_TIMEOUT).ifPresent(tiConf::setTimeout); | ||
configuration.getOptional(TIKV_GRPC_SCAN_TIMEOUT).ifPresent(tiConf::setScanTimeout); | ||
configuration | ||
.getOptional(TIKV_BATCH_GET_CONCURRENCY) | ||
.ifPresent(tiConf::setBatchGetConcurrency); | ||
configuration | ||
.getOptional(TIKV_BATCH_PUT_CONCURRENCY) | ||
.ifPresent(tiConf::setBatchPutConcurrency); | ||
configuration | ||
.getOptional(TIKV_BATCH_SCAN_CONCURRENCY) | ||
.ifPresent(tiConf::setBatchScanConcurrency); | ||
configuration | ||
.getOptional(TIKV_BATCH_DELETE_CONCURRENCY) | ||
.ifPresent(tiConf::setBatchDeleteConcurrency); | ||
return tiConf; | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We cannot list all valida TiKV parameters here, I think we should enable user to pass tikv.xx
options just like Flink Kafka Connector enable user to properties.xx
.
...r-tidb-cdc/src/main/java/com/ververica/cdc/connectors/tidb/table/TiDBTableSourceFactory.java
Outdated
Show resolved
Hide resolved
FactoryUtil.createTableFactoryHelper(this, context); | ||
|
||
final ReadableConfig config = helper.getOptions(); | ||
String hostname = config.get(HOSTNAME); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Option TDBSourceOptions.TIKV_PD_ADDRESSES
is required,but you do not add to TiDBTableSource
‘s constructor parameter.
.../java/com/ververica/cdc/connectors/tidb/table/RowDataTiKVEventDeserializationSchemaBase.java
Show resolved
Hide resolved
cherry-pick and squash commits, then force push. |
Thanks @eastfisher and @GOODBOY008 for the great work! Looks pretty good to me as our first version, I'll help organize the commits and merge them. |
This PR fixes issue #243.