You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "reswqa (via GitHub)" <gi...@apache.org> on 2023/07/04 11:42:04 UTC
[GitHub] [flink] reswqa commented on a diff in pull request #22933: [FLINK-32459][connector] Force set the parallelism of SocketTableSource to 1
reswqa commented on code in PR #22933:
URL: https://github.com/apache/flink/pull/22933#discussion_r1251913361
##########
flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/connectors/SocketDynamicTableSource.java:
##########
@@ -69,16 +73,28 @@ public ChangelogMode getChangelogMode() {
@Override
public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ return new DataStreamScanProvider() {
+ @Override
+ public DataStream<RowData> produceDataStream(
+ ProviderContext providerContext, StreamExecutionEnvironment execEnv) {
+ final DeserializationSchema<RowData> deserializer =
+ decodingFormat.createRuntimeDecoder(
+ runtimeProviderContext, producedDataType);
- // create runtime classes that are shipped to the cluster
+ final SocketSource socketSource =
+ new SocketSource(hostname, port, byteDelimiter, deserializer);
- final DeserializationSchema<RowData> deserializer =
- decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType);
+ return execEnv.fromSource(
+ socketSource, WatermarkStrategy.noWatermarks(), "SocketSource")
+ // SocketSource can only work with a parallelism of 1.
Review Comment:
Good sugesstion, added it.
##########
flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/connectors/SocketDynamicTableSource.java:
##########
@@ -69,16 +73,28 @@ public ChangelogMode getChangelogMode() {
@Override
public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ return new DataStreamScanProvider() {
+ @Override
+ public DataStream<RowData> produceDataStream(
+ ProviderContext providerContext, StreamExecutionEnvironment execEnv) {
+ final DeserializationSchema<RowData> deserializer =
+ decodingFormat.createRuntimeDecoder(
+ runtimeProviderContext, producedDataType);
- // create runtime classes that are shipped to the cluster
+ final SocketSource socketSource =
+ new SocketSource(hostname, port, byteDelimiter, deserializer);
- final DeserializationSchema<RowData> deserializer =
- decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType);
+ return execEnv.fromSource(
+ socketSource, WatermarkStrategy.noWatermarks(), "SocketSource")
+ // SocketSource can only work with a parallelism of 1.
Review Comment:
Good suggestion, added it.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org