You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/07/26 11:53:53 UTC
[GitHub] [flink] brachi-wernick opened a new pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
brachi-wernick opened a new pull request #16598:
URL: https://github.com/apache/flink/pull/16598
## What is the purpose of the change
I want to add Table API Support for GCP PubSub (Source Only for now)
Based on this document:
https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/sourcessinks/
Users will be able to run Flink SQL on PubSub.
## Brief change log
*(for example:)*
- added `PubSubTableSourceFactory` and `PubsubDynamicSource`
- json format support only
- options need to need to be set for the source: projectName and topic.
## Verifying this change
This change is already covered by existing tests, such as `PubSubSourceTest`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): yes (Flink dependencies with `provided` scope)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: don't know
- The runtime per-record code paths (performance sensitive): don't know
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? kind of
- If yes, how is the feature documented? hope you can direct me where should it be added
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7b7fb97cecd535742504992a9f884724bd9ed9ad Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 779cfe076236f909749dad31b5f16e5e17a8a29b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] brachi-wernick commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-930460899
@Airblader I'm ready from my side, anything else that should be addressed?
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
}, {
"hash" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388",
"triggerID" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"triggerType" : "PUSH"
}, {
"hash" : "4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24658",
"triggerID" : "4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24658)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28d9148e087008e819c73be7ac449c0ed050fd7a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811)
* 7f5e2749090dbdccf18a6a3ae4c899ac432808bb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347)
* 779cfe076236f909749dad31b5f16e5e17a8a29b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e6b0b05fadd9d6724a424c313a7b9617a854955f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259)
* ea804b7e9a1e7411605938da2e10f4bd95bfd883 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] Airblader commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r700835142
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,77 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ private static final ConfigOption<String> PROJECT_NAME =
Review comment:
I only meant to move `IDENTIFIER` back to the factory. The config options should still remain in a `@PublicEvolving` class `PubSubConnectorOptions`. Please move these back.
--
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
[GitHub] [flink] brachi-wernick commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r700523755
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
Review comment:
mistake (copy paste from another connector so forget to clean it 😬
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,67 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return PubSubConnectorConfigOptions.IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PubSubConnectorConfigOptions.PROJECT_NAME);
+ options.add(PubSubConnectorConfigOptions.SUBSCRIPTION);
+ options.add(FactoryUtil.FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // discover a suitable decoding format
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
+ helper.discoverDecodingFormat(
+ DeserializationFormatFactory.class, FactoryUtil.FORMAT);
+
+ // validate all options
+ helper.validate();
+
+ // get the validated options
+ final ReadableConfig options = helper.getOptions();
+ final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME);
+ final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION);
Review comment:
oops, sure
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink
+ * SQL dialect and the Flink Table API.
+ */
+@PublicEvolving
+public class PubSubConnectorConfigOptions {
+
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Name of the PubSub project backing this table.");
+
+ public static final ConfigOption<String> SUBSCRIPTION =
+ ConfigOptions.key("subscription")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Name of the PubSub subscription backing this table.");
+
+ public static final String IDENTIFIER = "pubsub";
Review comment:
So if it will be moved to the factory class, do we still need this option class? can I remove it? not sure i understand its purpose..
(For now I removed it lmk if it is needed)
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java
##########
@@ -0,0 +1,80 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** PubSub-backed {@link ScanTableSource}. */
+@Internal
+public class PubsubDynamicSource implements ScanTableSource {
+
+ private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class);
+
+ /** Name of the PubSub project backing this table. */
+ private final String project;
+ /** Name of the PubSub subscription backing this table. */
+ private final String subscription;
+ /** Scan format for decoding records from PubSub. */
+ private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
+ /** Data type that describes the final output of the source. */
+ private final DataType producedDataType;
+
+ public PubsubDynamicSource(
+ String project,
+ String subscription,
+ DecodingFormat<DeserializationSchema<RowData>> decodingFormat,
+ DataType producedDataType) {
+
+ this.project = project;
+ this.subscription = subscription;
+ this.decodingFormat = decodingFormat;
+ this.producedDataType = producedDataType;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ // create runtime classes that are shipped to the cluster
+
+ final DeserializationSchema<RowData> deserializer =
+ decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType);
+
+ try {
+ PubSubSource<RowData> source =
+ PubSubSource.newBuilder()
+ .withDeserializationSchema(deserializer)
+ .withProjectName(project)
+ .withSubscriptionName(subscription)
Review comment:
I think, most of the time, when you read from PubSub your cluster exists in GCP and you probably configure the state with GCS, so you already define the credentials.
Same for Kinesis, your cluster probably deployed in AWS with s3 state and you already define the credentials.(I can't see in Kinesis that they get credentials... maybe it should be addressed also there)
This can be relevant when you run in one cloud, let's say AWS ,and want to read data from GCP PubSub and vice versa.. or you run on premises cluster that reads data from kinesis/PubSub.. for those cases yes. we need to find a way to read credentials in some secured way, because tables can be done by external teams (self service)
--
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
[GitHub] [flink] Airblader commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r677386515
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
Review comment:
Can you please add JavaDoc to all classes? This is required for the CI to pass as well.
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
Review comment:
Please move config options to a separate `PubSubConnectorConfigOptions` class and mark it as `@PublicEvolving`, since config options are public API.
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
Review comment:
The naming convention for these so far would be `PubSubDynamicTableFactory`, can you please rename it?
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java
##########
@@ -0,0 +1,75 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** */
Review comment:
Please add proper documentation.
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java
##########
@@ -0,0 +1,75 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** */
+@Internal
+public class PubsubDynamicSource implements ScanTableSource {
+ private final String project;
+ private final String topic;
+ private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
+ private final DataType producedDataType;
+ private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class);
Review comment:
Can we move the static field to the top and separate it from instance fields?
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> FORMAT =
+ ConfigOptions.key("format").stringType().noDefaultValue();
+
+ @Override
+ public String factoryIdentifier() {
+ return "pubsub";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PROJECT_NAME);
+ options.add(TOPIC);
+ options.add(FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper utility
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // validate all options
+ helper.validateExcept("json.");
Review comment:
There are no `json.*` options? What is this doing?
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> FORMAT =
+ ConfigOptions.key("format").stringType().noDefaultValue();
+
+ @Override
+ public String factoryIdentifier() {
+ return "pubsub";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PROJECT_NAME);
+ options.add(TOPIC);
+ options.add(FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ // either implement your custom validation logic here ...
Review comment:
Looks like these comments should be removed.
--
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
[GitHub] [flink] brachi-wernick commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r680569542
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> FORMAT =
+ ConfigOptions.key("format").stringType().noDefaultValue();
+
+ @Override
+ public String factoryIdentifier() {
+ return "pubsub";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PROJECT_NAME);
+ options.add(TOPIC);
+ options.add(FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper utility
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // validate all options
+ helper.validateExcept("json.");
Review comment:
Got it, Thanks for the explanation!
I fixed it [in this revision](https://github.com/apache/flink/pull/16598/commits/e6b0b05fadd9d6724a424c313a7b9617a854955f)
--
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
[GitHub] [flink] brachi-wernick commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r677542021
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> FORMAT =
+ ConfigOptions.key("format").stringType().noDefaultValue();
+
+ @Override
+ public String factoryIdentifier() {
+ return "pubsub";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PROJECT_NAME);
+ options.add(TOPIC);
+ options.add(FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper utility
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // validate all options
+ helper.validateExcept("json.");
Review comment:
When there is json config in the create table sql, the validation here failed.
These are the json config I see:
```
-- 'json.map-null-key.mode' = 'FAIL', -- Optional flag to control the handling mode when serializing null key for map data, FAIL by default. Option DROP will drop null key entries for map data. Option LITERAL will use 'map-null-key.literal' as key literal.
-- 'json.encode.decimal-as-plain-number' = 'false', -- Optional flag to specify whether to encode all decimals as plain numbers instead of possible scientific notations, false by default.
-- 'json.map-null-key.literal' = 'null', -- Optional flag to specify string literal for null keys when 'map-null-key.mode' is LITERAL, "null" by default.
-- 'json.fail-on-missing-field' = 'false', -- Optional flag to specify whether to fail if a field is missing or not, false by default.
-- 'json.ignore-parse-errors' = 'false', -- Optional flag to skip fields and rows with parse errors instead of failing; fields are set to null in case of errors, false by default.
-- 'json.timestamp-format.standard' = 'SQL', -- Optional flag to specify timestamp format, SQL by default. Option ISO-8601 will parse input timestamp in "yyyy-MM-ddTHH:mm:ss.s{precision}" format and output timestamp in the same format. Option SQL will parse input timestamp in "yyyy-MM-dd HH:mm:ss.s{precision}" format and output timestamp in the same format.
```
Any other idea how to address 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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 263952ee9a0288ec672d5776ebaa396dbf776bd6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 263952ee9a0288ec672d5776ebaa396dbf776bd6 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ea804b7e9a1e7411605938da2e10f4bd95bfd883 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] brachi-wernick commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-909556618
@Airblader I'm ready from my side, anything else that should be addressed?
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
}, {
"hash" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388",
"triggerID" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"triggerType" : "PUSH"
}, {
"hash" : "4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb8bdc31173db9b16a82fbf596334272193da91c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388)
* 4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] brachi-wernick commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-887745257
@Airblader I address most of the comment, and reply when needed.
> We'll also need to update the documentation to document the PubSub connector for Table API.
Should I just add `flink/docs/content/docs/connectors/table/pubsub.md` file?
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 65f256e785b3ac499d38eee63d177f6a6059b609 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052)
* e6b0b05fadd9d6724a424c313a7b9617a854955f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] Airblader commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-889715554
@brachi-wernick Yeah, that would be great.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28d9148e087008e819c73be7ac449c0ed050fd7a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811)
* 7f5e2749090dbdccf18a6a3ae4c899ac432808bb UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
}, {
"hash" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388",
"triggerID" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb8bdc31173db9b16a82fbf596334272193da91c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] brachi-wernick commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-909556618
@Airblader I'm ready from my side, anything else that should be addressed?
--
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
[GitHub] [flink] brachi-wernick commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-930460899
@Airblader I'm ready from my side, anything else that should be addressed?
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 263952ee9a0288ec672d5776ebaa396dbf776bd6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981)
* 7b7fb97cecd535742504992a9f884724bd9ed9ad UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ea804b7e9a1e7411605938da2e10f4bd95bfd883 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417)
* 28d9148e087008e819c73be7ac449c0ed050fd7a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] Airblader commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r679732276
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> FORMAT =
+ ConfigOptions.key("format").stringType().noDefaultValue();
+
+ @Override
+ public String factoryIdentifier() {
+ return "pubsub";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PROJECT_NAME);
+ options.add(TOPIC);
+ options.add(FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper utility
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // validate all options
+ helper.validateExcept("json.");
Review comment:
Can you post the full DDL you're using that causes errors? When you specify `'format' = 'json'`, the respective format options should be considered automatically.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28d9148e087008e819c73be7ac449c0ed050fd7a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 65f256e785b3ac499d38eee63d177f6a6059b609 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
}, {
"hash" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388",
"triggerID" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"triggerType" : "PUSH"
}, {
"hash" : "4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24658",
"triggerID" : "4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4ddd79c8bd25ed27a74bcc2fb071e57a1ac7093a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24658)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
}, {
"hash" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 779cfe076236f909749dad31b5f16e5e17a8a29b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352)
* eb8bdc31173db9b16a82fbf596334272193da91c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e6b0b05fadd9d6724a424c313a7b9617a854955f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259)
* ea804b7e9a1e7411605938da2e10f4bd95bfd883 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7b7fb97cecd535742504992a9f884724bd9ed9ad Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039)
* 65f256e785b3ac499d38eee63d177f6a6059b609 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] brachi-wernick commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r680498203
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> FORMAT =
+ ConfigOptions.key("format").stringType().noDefaultValue();
+
+ @Override
+ public String factoryIdentifier() {
+ return "pubsub";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PROJECT_NAME);
+ options.add(TOPIC);
+ options.add(FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper utility
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // validate all options
+ helper.validateExcept("json.");
Review comment:
The DDL:
```
CREATE TABLE `vvp`.`default`.`my_table` (
`user_id` VARCHAR(2147483647),
`country_code` VARCHAR(2147483647),
`event` VARCHAR(2147483647),
`os` VARCHAR(2147483647)
)
COMMENT ''
WITH (
'connector' = 'pubsub',
'format' = 'json',
'json.map-null-key.literal' = 'null',
'json.map-null-key.mode' = 'LITERAL',
'projectName' = 'myProj,
'topic' = 'mySubscription'
);
```
The error I get (if I don't exclude the 'json' from the validation):
```
Invalid connector configuration
The source connector for table 'vvp.default.my_table' could not be created. Configured properties are: ['connector'='pubsub', 'format'='json', 'json.map-null-key.literal'='null', 'json.map-null-key.mode'='LITERAL', 'projectName'='myProj', 'topic'='mySubscription'].
Cause: Unsupported options found for 'pubsub'.
Unsupported options:
json.map-null-key.literal
json.map-null-key.mode
```
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e6b0b05fadd9d6724a424c313a7b9617a854955f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] martinbomio commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
martinbomio commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-1075488768
Hello! What's the status of this PR, we have an internal version of this table connector and would love to get rid of it/ contribute 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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ea804b7e9a1e7411605938da2e10f4bd95bfd883 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417)
* 28d9148e087008e819c73be7ac449c0ed050fd7a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] Airblader commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r680563823
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+ // define all options statically
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> FORMAT =
+ ConfigOptions.key("format").stringType().noDefaultValue();
+
+ @Override
+ public String factoryIdentifier() {
+ return "pubsub";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PROJECT_NAME);
+ options.add(TOPIC);
+ options.add(FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper utility
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // validate all options
+ helper.validateExcept("json.");
Review comment:
Thanks for sharing that information. We cannot just exclude `json.*` because if you use another format, the format options would also have a different prefix.
The reason this fails is because we're currently calling things in the wrong order. You can move the call to `helper.discoverDecodingFormat` to before (what will now be) `helper.validate()`. This will register the options of the discovered format internally so they're ignored when validating the connector options.
--
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
[GitHub] [flink] Airblader commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-887465991
We'll also need to update the documentation to document the PubSub connector for Table API.
--
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
[GitHub] [flink] flinkbot commented on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886635957
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 263952ee9a0288ec672d5776ebaa396dbf776bd6 (Mon Jul 26 11:56:01 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-23501).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 65f256e785b3ac499d38eee63d177f6a6059b609 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052)
* e6b0b05fadd9d6724a424c313a7b9617a854955f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7f5e2749090dbdccf18a6a3ae4c899ac432808bb Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347)
* 779cfe076236f909749dad31b5f16e5e17a8a29b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] Airblader commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r719102620
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink
+ * SQL dialect and the Flink Table API.
+ */
+@PublicEvolving
+public class PubSubConnectorConfigOptions {
Review comment:
@brachi-wernick This comment is still open, isn't 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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 28d9148e087008e819c73be7ac449c0ed050fd7a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811)
* 7f5e2749090dbdccf18a6a3ae4c899ac432808bb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347)
* 779cfe076236f909749dad31b5f16e5e17a8a29b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16598:
URL: https://github.com/apache/flink/pull/16598#issuecomment-886655820
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20981",
"triggerID" : "263952ee9a0288ec672d5776ebaa396dbf776bd6",
"triggerType" : "PUSH"
}, {
"hash" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21039",
"triggerID" : "7b7fb97cecd535742504992a9f884724bd9ed9ad",
"triggerType" : "PUSH"
}, {
"hash" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21052",
"triggerID" : "65f256e785b3ac499d38eee63d177f6a6059b609",
"triggerType" : "PUSH"
}, {
"hash" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21259",
"triggerID" : "e6b0b05fadd9d6724a424c313a7b9617a854955f",
"triggerType" : "PUSH"
}, {
"hash" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21417",
"triggerID" : "ea804b7e9a1e7411605938da2e10f4bd95bfd883",
"triggerType" : "PUSH"
}, {
"hash" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=22811",
"triggerID" : "28d9148e087008e819c73be7ac449c0ed050fd7a",
"triggerType" : "PUSH"
}, {
"hash" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23347",
"triggerID" : "7f5e2749090dbdccf18a6a3ae4c899ac432808bb",
"triggerType" : "PUSH"
}, {
"hash" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352",
"triggerID" : "779cfe076236f909749dad31b5f16e5e17a8a29b",
"triggerType" : "PUSH"
}, {
"hash" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388",
"triggerID" : "eb8bdc31173db9b16a82fbf596334272193da91c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 779cfe076236f909749dad31b5f16e5e17a8a29b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23352)
* eb8bdc31173db9b16a82fbf596334272193da91c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23388)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] Airblader commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r700234951
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
+
+This connector provides a Source that can read from Google Cloud PubSub.
Review comment:
```suggestion
This connector allows reading from Google Cloud PubSub.
```
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
+
+This connector provides a Source that can read from Google Cloud PubSub.
+
+Dependencies
+------------
+
+{{< sql_download_table "pubsub" >}}
+
+The PubSub connector is not currently part of the binary distribution.
+See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+How to create a PubSub table
+----------------
+
+The example below shows how to create a PubSub table:
+
+```sql
+CREATE TABLE PubSubTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `behavior` STRING
+) WITH (
+ 'connector' = 'pubsub',
+ 'format' = 'json',
+ 'projectName' = 'gcpProject',
+ 'subscription' = 'mySubscription'
+)
+```
+
+Available Metadata
+------------------
+
+The following connector metadata can be accessed as metadata columns in a table definition.
+
+The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`).
+Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation.
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Key</th>
+ <th class="text-center" style="width: 30%">Data Type</th>
+ <th class="text-center" style="width: 40%">Description</th>
+ <th class="text-center" style="width: 5%">R/W</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>projectName</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the project in GCP contains the PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ <tr>
+ <td><code>subscription</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the GCP PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ </tbody>
+</table>
+
+
+Connector Options
+----------------
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-center" style="width: 8%">Required</th>
+ <th class="text-center" style="width: 7%">Default</th>
+ <th class="text-center" style="width: 10%">Type</th>
+ <th class="text-center" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>connector</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify what connector to use, for PubSub use <code>'pubsub'</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>projectName</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Project name in GCP where the PubSub subscription/topic is defined.</td>
+ </tr>
+ <tr>
+ <td><h5>subscription</h5></td>
+ <td>required by source</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>PubSub subscriptiob name to read data from when the table is used as source.</td>
Review comment:
```suggestion
<td>Name of the PubSub subscription to read from.</td>
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink
+ * SQL dialect and the Flink Table API.
+ */
+@PublicEvolving
+public class PubSubConnectorConfigOptions {
Review comment:
Can we please rename this to `PubSubConnectorOptions`? We recently went through some changes and want to make this consistent.
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,67 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return PubSubConnectorConfigOptions.IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PubSubConnectorConfigOptions.PROJECT_NAME);
+ options.add(PubSubConnectorConfigOptions.SUBSCRIPTION);
+ options.add(FactoryUtil.FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // discover a suitable decoding format
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
+ helper.discoverDecodingFormat(
+ DeserializationFormatFactory.class, FactoryUtil.FORMAT);
+
+ // validate all options
Review comment:
```suggestion
```
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
+
+This connector provides a Source that can read from Google Cloud PubSub.
+
+Dependencies
+------------
+
+{{< sql_download_table "pubsub" >}}
+
+The PubSub connector is not currently part of the binary distribution.
+See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+How to create a PubSub table
+----------------
+
+The example below shows how to create a PubSub table:
+
+```sql
+CREATE TABLE PubSubTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `behavior` STRING
+) WITH (
+ 'connector' = 'pubsub',
+ 'format' = 'json',
+ 'projectName' = 'gcpProject',
+ 'subscription' = 'mySubscription'
+)
+```
+
+Available Metadata
+------------------
+
+The following connector metadata can be accessed as metadata columns in a table definition.
+
+The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`).
+Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation.
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Key</th>
+ <th class="text-center" style="width: 30%">Data Type</th>
+ <th class="text-center" style="width: 40%">Description</th>
+ <th class="text-center" style="width: 5%">R/W</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>projectName</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the project in GCP contains the PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ <tr>
+ <td><code>subscription</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the GCP PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ </tbody>
+</table>
+
+
+Connector Options
+----------------
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-center" style="width: 8%">Required</th>
+ <th class="text-center" style="width: 7%">Default</th>
+ <th class="text-center" style="width: 10%">Type</th>
+ <th class="text-center" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>connector</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify what connector to use, for PubSub use <code>'pubsub'</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>projectName</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Project name in GCP where the PubSub subscription/topic is defined.</td>
+ </tr>
+ <tr>
+ <td><h5>subscription</h5></td>
+ <td>required by source</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>PubSub subscriptiob name to read data from when the table is used as source.</td>
+ </tr>
+ <tr>
+ <td><h5>format</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The format used to deserialize and serialize the value part of PubSub messages.
+ Please refer to the <a href="{{< ref "docs/connectors/table/formats/overview" >}}">formats</a> page for
+ more details and more format options.
+ </td>
+ </tr>
+
+ </tbody>
+</table>
+
+Features
+----------------
+ N/A
Review comment:
This empty section doesn't seem useful.
```suggestion
```
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
Review comment:
What's the purpose of this label here given that this is a source-only connector?
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
+
+This connector provides a Source that can read from Google Cloud PubSub.
+
+Dependencies
+------------
+
+{{< sql_download_table "pubsub" >}}
+
+The PubSub connector is not currently part of the binary distribution.
Review comment:
I think if we offer it, we should also distribute it? CC @twalthr
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink
+ * SQL dialect and the Flink Table API.
+ */
Review comment:
```suggestion
/** Options for the PubSub connector. */
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink
+ * SQL dialect and the Flink Table API.
+ */
+@PublicEvolving
+public class PubSubConnectorConfigOptions {
+
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Name of the PubSub project backing this table.");
+
+ public static final ConfigOption<String> SUBSCRIPTION =
+ ConfigOptions.key("subscription")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Name of the PubSub subscription backing this table.");
+
+ public static final String IDENTIFIER = "pubsub";
+}
Review comment:
```suggestion
private PubSubConnectorConfigOptions() {}
}
```
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
+
+This connector provides a Source that can read from Google Cloud PubSub.
+
+Dependencies
+------------
+
+{{< sql_download_table "pubsub" >}}
+
+The PubSub connector is not currently part of the binary distribution.
+See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+How to create a PubSub table
+----------------
+
+The example below shows how to create a PubSub table:
+
+```sql
+CREATE TABLE PubSubTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `behavior` STRING
+) WITH (
+ 'connector' = 'pubsub',
+ 'format' = 'json',
+ 'projectName' = 'gcpProject',
+ 'subscription' = 'mySubscription'
+)
+```
+
+Available Metadata
+------------------
+
+The following connector metadata can be accessed as metadata columns in a table definition.
+
+The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`).
+Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation.
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Key</th>
+ <th class="text-center" style="width: 30%">Data Type</th>
+ <th class="text-center" style="width: 40%">Description</th>
+ <th class="text-center" style="width: 5%">R/W</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>projectName</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the project in GCP contains the PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ <tr>
+ <td><code>subscription</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the GCP PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ </tbody>
+</table>
+
+
+Connector Options
+----------------
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-center" style="width: 8%">Required</th>
+ <th class="text-center" style="width: 7%">Default</th>
+ <th class="text-center" style="width: 10%">Type</th>
+ <th class="text-center" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>connector</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify what connector to use, for PubSub use <code>'pubsub'</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>projectName</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Project name in GCP where the PubSub subscription/topic is defined.</td>
Review comment:
```suggestion
<td>Name of the GCP project that contains the subscription.</td>
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,67 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return PubSubConnectorConfigOptions.IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PubSubConnectorConfigOptions.PROJECT_NAME);
+ options.add(PubSubConnectorConfigOptions.SUBSCRIPTION);
+ options.add(FactoryUtil.FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // discover a suitable decoding format
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
+ helper.discoverDecodingFormat(
+ DeserializationFormatFactory.class, FactoryUtil.FORMAT);
+
+ // validate all options
+ helper.validate();
+
+ // get the validated options
Review comment:
```suggestion
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink
+ * SQL dialect and the Flink Table API.
+ */
+@PublicEvolving
+public class PubSubConnectorConfigOptions {
+
+ public static final ConfigOption<String> PROJECT_NAME =
+ ConfigOptions.key("projectName")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Name of the PubSub project backing this table.");
+
+ public static final ConfigOption<String> SUBSCRIPTION =
+ ConfigOptions.key("subscription")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Name of the PubSub subscription backing this table.");
+
+ public static final String IDENTIFIER = "pubsub";
Review comment:
This isn't a config option. We currently have these reside in the factory class. I don't think that's ideal (since it is internal), but for now we should move this one to the factory as well, I think.
##########
File path: docs/content/docs/connectors/table/pubsub.md
##########
@@ -0,0 +1,145 @@
+---
+title: PubSub
+weight: 20
+type: docs
+aliases:
+ - /dev/table/connectors/pubsub.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Google Cloud PubSub SQL Connector
+
+{{< label "Scan Source: Unbounded" >}}
+{{< label "Sink: Streaming Append Mode" >}}
+
+This connector provides a Source that can read from Google Cloud PubSub.
+
+Dependencies
+------------
+
+{{< sql_download_table "pubsub" >}}
+
+The PubSub connector is not currently part of the binary distribution.
+See how to link with it for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+How to create a PubSub table
+----------------
+
+The example below shows how to create a PubSub table:
+
+```sql
+CREATE TABLE PubSubTable (
+ `user_id` BIGINT,
+ `item_id` BIGINT,
+ `behavior` STRING
+) WITH (
+ 'connector' = 'pubsub',
+ 'format' = 'json',
+ 'projectName' = 'gcpProject',
+ 'subscription' = 'mySubscription'
+)
+```
+
+Available Metadata
+------------------
+
+The following connector metadata can be accessed as metadata columns in a table definition.
+
+The `R/W` column defines whether a metadata field is readable (`R`) and/or writable (`W`).
+Read-only columns must be declared `VIRTUAL` to exclude them during an `INSERT INTO` operation.
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Key</th>
+ <th class="text-center" style="width: 30%">Data Type</th>
+ <th class="text-center" style="width: 40%">Description</th>
+ <th class="text-center" style="width: 5%">R/W</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>projectName</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the project in GCP contains the PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ <tr>
+ <td><code>subscription</code></td>
+ <td><code>STRING NOT NULL</code></td>
+ <td>Name of the GCP PubSub subscription.</td>
+ <td><code>R</code></td>
+ </tr>
+ </tbody>
+</table>
Review comment:
The source does not implement any metadata, so this entire section should be removed.
```suggestion
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,67 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return PubSubConnectorConfigOptions.IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PubSubConnectorConfigOptions.PROJECT_NAME);
+ options.add(PubSubConnectorConfigOptions.SUBSCRIPTION);
+ options.add(FactoryUtil.FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // discover a suitable decoding format
Review comment:
All of these comments add no real value as the code reads clear enough.
```suggestion
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java
##########
@@ -0,0 +1,80 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** PubSub-backed {@link ScanTableSource}. */
+@Internal
+public class PubsubDynamicSource implements ScanTableSource {
+
+ private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class);
+
+ /** Name of the PubSub project backing this table. */
+ private final String project;
+ /** Name of the PubSub subscription backing this table. */
+ private final String subscription;
+ /** Scan format for decoding records from PubSub. */
+ private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
+ /** Data type that describes the final output of the source. */
+ private final DataType producedDataType;
+
+ public PubsubDynamicSource(
+ String project,
+ String subscription,
+ DecodingFormat<DeserializationSchema<RowData>> decodingFormat,
+ DataType producedDataType) {
+
+ this.project = project;
+ this.subscription = subscription;
+ this.decodingFormat = decodingFormat;
+ this.producedDataType = producedDataType;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ // create runtime classes that are shipped to the cluster
+
+ final DeserializationSchema<RowData> deserializer =
+ decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType);
+
+ try {
+ PubSubSource<RowData> source =
+ PubSubSource.newBuilder()
+ .withDeserializationSchema(deserializer)
+ .withProjectName(project)
+ .withSubscriptionName(subscription)
Review comment:
(We'll need to think about how to provide credentials (not through environment variables) as well, I think, but as a first version we can leave this as is.)
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,67 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return PubSubConnectorConfigOptions.IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PubSubConnectorConfigOptions.PROJECT_NAME);
+ options.add(PubSubConnectorConfigOptions.SUBSCRIPTION);
+ options.add(FactoryUtil.FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // discover a suitable decoding format
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
+ helper.discoverDecodingFormat(
+ DeserializationFormatFactory.class, FactoryUtil.FORMAT);
+
+ // validate all options
+ helper.validate();
+
+ // get the validated options
+ final ReadableConfig options = helper.getOptions();
+ final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME);
+ final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION);
+
+ // derive the produced data type (excluding computed columns) from the catalog table
+ final DataType producedDataType =
+ context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType();
+
+ // create and return dynamic table source
Review comment:
```suggestion
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,67 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return PubSubConnectorConfigOptions.IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PubSubConnectorConfigOptions.PROJECT_NAME);
+ options.add(PubSubConnectorConfigOptions.SUBSCRIPTION);
+ options.add(FactoryUtil.FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // discover a suitable decoding format
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
+ helper.discoverDecodingFormat(
+ DeserializationFormatFactory.class, FactoryUtil.FORMAT);
+
+ // validate all options
+ helper.validate();
+
+ // get the validated options
+ final ReadableConfig options = helper.getOptions();
+ final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME);
+ final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION);
+
+ // derive the produced data type (excluding computed columns) from the catalog table
Review comment:
```suggestion
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java
##########
@@ -0,0 +1,67 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Factory for creating {@link PubsubDynamicSource}. */
+@Internal
+public class PubSubDynamicTableFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return PubSubConnectorConfigOptions.IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(PubSubConnectorConfigOptions.PROJECT_NAME);
+ options.add(PubSubConnectorConfigOptions.SUBSCRIPTION);
+ options.add(FactoryUtil.FORMAT);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ // discover a suitable decoding format
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
+ helper.discoverDecodingFormat(
+ DeserializationFormatFactory.class, FactoryUtil.FORMAT);
+
+ // validate all options
+ helper.validate();
+
+ // get the validated options
+ final ReadableConfig options = helper.getOptions();
+ final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME);
+ final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION);
Review comment:
Rename to `subscription`?
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java
##########
@@ -0,0 +1,80 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** PubSub-backed {@link ScanTableSource}. */
+@Internal
+public class PubsubDynamicSource implements ScanTableSource {
+
+ private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class);
+
+ /** Name of the PubSub project backing this table. */
+ private final String project;
+ /** Name of the PubSub subscription backing this table. */
+ private final String subscription;
+ /** Scan format for decoding records from PubSub. */
+ private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
+ /** Data type that describes the final output of the source. */
+ private final DataType producedDataType;
+
+ public PubsubDynamicSource(
+ String project,
+ String subscription,
+ DecodingFormat<DeserializationSchema<RowData>> decodingFormat,
+ DataType producedDataType) {
+
+ this.project = project;
+ this.subscription = subscription;
+ this.decodingFormat = decodingFormat;
+ this.producedDataType = producedDataType;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ // create runtime classes that are shipped to the cluster
+
Review comment:
```suggestion
```
##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java
##########
@@ -0,0 +1,80 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** PubSub-backed {@link ScanTableSource}. */
+@Internal
+public class PubsubDynamicSource implements ScanTableSource {
+
+ private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class);
+
+ /** Name of the PubSub project backing this table. */
+ private final String project;
+ /** Name of the PubSub subscription backing this table. */
+ private final String subscription;
+ /** Scan format for decoding records from PubSub. */
+ private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
+ /** Data type that describes the final output of the source. */
+ private final DataType producedDataType;
+
+ public PubsubDynamicSource(
+ String project,
+ String subscription,
+ DecodingFormat<DeserializationSchema<RowData>> decodingFormat,
+ DataType producedDataType) {
+
+ this.project = project;
+ this.subscription = subscription;
+ this.decodingFormat = decodingFormat;
+ this.producedDataType = producedDataType;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ // create runtime classes that are shipped to the cluster
+
+ final DeserializationSchema<RowData> deserializer =
+ decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType);
+
+ try {
+ PubSubSource<RowData> source =
+ PubSubSource.newBuilder()
+ .withDeserializationSchema(deserializer)
+ .withProjectName(project)
+ .withSubscriptionName(subscription)
+ .build();
+ return SourceFunctionProvider.of(source, false);
+ } catch (IOException e) {
+ throw new RuntimeException("failed to create source", e);
Review comment:
```suggestion
throw new RuntimeException("Failed to create PubSub source.", e);
```
--
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