You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "tisonkun (via GitHub)" <gi...@apache.org> on 2023/07/24 06:03:18 UTC
[GitHub] [flink-connector-pulsar] tisonkun opened a new pull request, #56: [FLINK-XXXXX] Basic table factory for Pulsar connector
tisonkun opened a new pull request, #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56
This patch consists work from multiple contributors. I'll respect all their credits before merging.
## Significant changes
*(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for
convenience.)*
- [ ] Dependencies have been added or upgraded
- [ ] Public API has been changed (Public API is any class annotated with `@Public(Evolving)`)
- [ ] Serializers have been changed
- [ ] New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304163565
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableValidationUtils.java:
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.RowKind;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.shade.com.google.common.collect.Sets;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FIELDS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_CUSTOM_TOPIC_ROUTER;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_TOPIC_ROUTING_MODE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AFTER_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.TOPICS;
+import static org.apache.pulsar.common.naming.TopicName.isValid;
+
+/** Util class for source and sink validation rules. */
+public class PulsarTableValidationUtils {
+
+ private PulsarTableValidationUtils() {}
+
+ public static void validatePrimaryKeyConstraints(
+ ObjectIdentifier tableName,
+ int[] primaryKeyIndexes,
+ FactoryUtil.TableFactoryHelper helper) {
+ final DecodingFormat<DeserializationSchema<RowData>> format =
+ getValueDecodingFormat(helper);
+ if (primaryKeyIndexes.length > 0
+ && format.getChangelogMode().containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ String.format(
+ "The Pulsar table '%s' with '%s' format doesn't support defining PRIMARY KEY constraint"
+ + " on the table, because it can't guarantee the semantic of primary key.",
+ tableName.asSummaryString(), format));
+ }
+ }
+
+ public static void validateTableSourceOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateStartCursorConfigs(tableOptions);
+ validateStopCursorConfigs(tableOptions);
+ validateSubscriptionTypeConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ }
+
+ public static void validateTableSinkOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ validateSinkRoutingConfigs(tableOptions);
+ }
+
+ protected static void validateTopicsConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.get(TOPICS).isEmpty()) {
+ throw new ValidationException("The topics list should not be empty.");
+ }
+
+ for (String topic : tableOptions.get(TOPICS)) {
+ if (!isValid(topic)) {
+ throw new ValidationException(
+ String.format("The topics name %s is not a valid topic name.", topic));
+ }
+ }
+ }
+
+ protected static void validateStartCursorConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SOURCE_START_FROM_MESSAGE_ID).isPresent()
+ && tableOptions.getOptional(SOURCE_START_FROM_PUBLISH_TIME).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SOURCE_START_FROM_MESSAGE_ID, SOURCE_START_FROM_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateStopCursorConfigs(ReadableConfig tableOptions) {
+ Set<ConfigOption<?>> conflictConfigOptions =
+ Sets.newHashSet(
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME);
+
+ long configsNums =
+ conflictConfigOptions.stream()
+ .map(tableOptions::getOptional)
+ .filter(Optional::isPresent)
+ .count();
+
+ if (configsNums > 1) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s, %s and %s can be specified. Detected more than 1 of them",
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateSubscriptionTypeConfigs(ReadableConfig tableOptions) {
+ SubscriptionType subscriptionType = tableOptions.get(SOURCE_SUBSCRIPTION_TYPE);
+
+ if (subscriptionType == SubscriptionType.Failover) {
+ throw new ValidationException(
+ String.format(
+ "%s SubscriptionType is not supported. ", SubscriptionType.Failover));
+ }
+ }
+
+ protected static void validateKeyFormatConfigs(ReadableConfig tableOptions) {
+ final Optional<String> optionalKeyFormat = tableOptions.getOptional(KEY_FORMAT);
+ final Optional<List<String>> optionalKeyFields = tableOptions.getOptional(KEY_FIELDS);
+ if (!optionalKeyFormat.isPresent() && optionalKeyFields.isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "The option '%s' can only be declared if a key format is defined using '%s'.",
+ KEY_FIELDS.key(), KEY_FORMAT.key()));
+ } else if (optionalKeyFormat.isPresent()
+ && (!optionalKeyFields.isPresent() || optionalKeyFields.get().size() == 0)) {
+ throw new ValidationException(
+ String.format(
+ "A key format '%s' requires the declaration of one or more of key fields using '%s'.",
+ KEY_FORMAT.key(), KEY_FIELDS.key()));
+ }
+ }
+
+ protected static void validateSinkRoutingConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).orElse(TopicRoutingMode.ROUND_ROBIN)
+ == TopicRoutingMode.CUSTOM) {
+ throw new ValidationException(
+ String.format(
+ "Only %s and %s can be used. For %s, please use sink.custom-topic-router for"
+ + "custom topic router and do not set this config.",
+ TopicRoutingMode.ROUND_ROBIN,
+ TopicRoutingMode.MESSAGE_KEY_HASH,
+ TopicRoutingMode.CUSTOM));
+ }
+ if (tableOptions.getOptional(SINK_CUSTOM_TOPIC_ROUTER).isPresent()
+ && tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SINK_CUSTOM_TOPIC_ROUTER, SINK_TOPIC_ROUTING_MODE));
+ }
+ }
+
+ protected static void validateUpsertModeKeyConstraints(
+ ReadableConfig tableOptions, int[] primaryKeyIndexes) {
+ if (!tableOptions.getOptional(KEY_FIELDS).isPresent()) {
+ throw new ValidationException(
+ "Upsert mode requires key.fields set to the primary key fields, should be set");
+ }
+
+ if (tableOptions.getOptional(KEY_FIELDS).get().size() == 0
+ || primaryKeyIndexes.length == 0) {
+ throw new ValidationException(
+ "'upsert-pulsar' require to define a PRIMARY KEY constraint. "
Review Comment:
Yes. I'm planning to add the upsert mode in a following PR.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchema.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.api.common.functions.util.ListCollector;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+
+import org.apache.pulsar.client.api.Message;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A specific {@link PulsarDeserializationSchema} for {@link PulsarTableSource}.
+ *
+ * <p>Both Flink's key decoding format and value decoding format are wrapped in this class. It is
+ * responsible for getting metadata fields from a physical pulsar message body, and the final
+ * projection mapping from Pulsar message fields to Flink row.
+ *
+ * <p>After retrieving key and value bytes and convert them into a list of {@link RowData}, it then
+ * delegates metadata appending, key and value {@link RowData} combining to a {@link
+ * PulsarRowDataConverter} instance.
+ */
+public class PulsarTableDeserializationSchema implements PulsarDeserializationSchema<RowData> {
+ private static final long serialVersionUID = -3298784447432136216L;
Review Comment:
Fixed.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304162722
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarWritableMetadata.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.sink;
+
+import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessageBuilder;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.bouncycastle.util.Arrays;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Stream;
+
+/** A class used to manage WritableMetadata for Pulsar SQL sink connector. */
+public class PulsarWritableMetadata implements Serializable {
+
+ private static final long serialVersionUID = 8117156158379846715L;
Review Comment:
Fixed.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304184103
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
Review Comment:
They are totally different concepts. You may think of Flink deployment type can be default to application mode, but there is no reason for a default deployment name.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304193141
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableValidationUtils.java:
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.RowKind;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.shade.com.google.common.collect.Sets;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FIELDS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_CUSTOM_TOPIC_ROUTER;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_TOPIC_ROUTING_MODE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AFTER_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.TOPICS;
+import static org.apache.pulsar.common.naming.TopicName.isValid;
+
+/** Util class for source and sink validation rules. */
+public class PulsarTableValidationUtils {
+
+ private PulsarTableValidationUtils() {}
+
+ public static void validatePrimaryKeyConstraints(
+ ObjectIdentifier tableName,
+ int[] primaryKeyIndexes,
+ FactoryUtil.TableFactoryHelper helper) {
+ final DecodingFormat<DeserializationSchema<RowData>> format =
+ getValueDecodingFormat(helper);
+ if (primaryKeyIndexes.length > 0
+ && format.getChangelogMode().containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ String.format(
+ "The Pulsar table '%s' with '%s' format doesn't support defining PRIMARY KEY constraint"
+ + " on the table, because it can't guarantee the semantic of primary key.",
+ tableName.asSummaryString(), format));
+ }
+ }
+
+ public static void validateTableSourceOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateStartCursorConfigs(tableOptions);
+ validateStopCursorConfigs(tableOptions);
+ validateSubscriptionTypeConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ }
+
+ public static void validateTableSinkOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ validateSinkRoutingConfigs(tableOptions);
+ }
+
+ protected static void validateTopicsConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.get(TOPICS).isEmpty()) {
+ throw new ValidationException("The topics list should not be empty.");
+ }
+
+ for (String topic : tableOptions.get(TOPICS)) {
+ if (!isValid(topic)) {
+ throw new ValidationException(
+ String.format("The topics name %s is not a valid topic name.", topic));
+ }
+ }
+ }
+
+ protected static void validateStartCursorConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SOURCE_START_FROM_MESSAGE_ID).isPresent()
+ && tableOptions.getOptional(SOURCE_START_FROM_PUBLISH_TIME).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SOURCE_START_FROM_MESSAGE_ID, SOURCE_START_FROM_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateStopCursorConfigs(ReadableConfig tableOptions) {
+ Set<ConfigOption<?>> conflictConfigOptions =
+ Sets.newHashSet(
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME);
+
+ long configsNums =
+ conflictConfigOptions.stream()
+ .map(tableOptions::getOptional)
+ .filter(Optional::isPresent)
+ .count();
+
+ if (configsNums > 1) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s, %s and %s can be specified. Detected more than 1 of them",
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateSubscriptionTypeConfigs(ReadableConfig tableOptions) {
+ SubscriptionType subscriptionType = tableOptions.get(SOURCE_SUBSCRIPTION_TYPE);
+
+ if (subscriptionType == SubscriptionType.Failover) {
+ throw new ValidationException(
+ String.format(
+ "%s SubscriptionType is not supported. ", SubscriptionType.Failover));
+ }
+ }
+
+ protected static void validateKeyFormatConfigs(ReadableConfig tableOptions) {
+ final Optional<String> optionalKeyFormat = tableOptions.getOptional(KEY_FORMAT);
+ final Optional<List<String>> optionalKeyFields = tableOptions.getOptional(KEY_FIELDS);
+ if (!optionalKeyFormat.isPresent() && optionalKeyFields.isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "The option '%s' can only be declared if a key format is defined using '%s'.",
+ KEY_FIELDS.key(), KEY_FORMAT.key()));
+ } else if (optionalKeyFormat.isPresent()
+ && (!optionalKeyFields.isPresent() || optionalKeyFields.get().size() == 0)) {
+ throw new ValidationException(
+ String.format(
+ "A key format '%s' requires the declaration of one or more of key fields using '%s'.",
+ KEY_FORMAT.key(), KEY_FIELDS.key()));
+ }
+ }
+
+ protected static void validateSinkRoutingConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).orElse(TopicRoutingMode.ROUND_ROBIN)
+ == TopicRoutingMode.CUSTOM) {
+ throw new ValidationException(
+ String.format(
+ "Only %s and %s can be used. For %s, please use sink.custom-topic-router for"
+ + "custom topic router and do not set this config.",
+ TopicRoutingMode.ROUND_ROBIN,
+ TopicRoutingMode.MESSAGE_KEY_HASH,
+ TopicRoutingMode.CUSTOM));
+ }
+ if (tableOptions.getOptional(SINK_CUSTOM_TOPIC_ROUTER).isPresent()
+ && tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SINK_CUSTOM_TOPIC_ROUTER, SINK_TOPIC_ROUTING_MODE));
+ }
+ }
+
+ protected static void validateUpsertModeKeyConstraints(
+ ReadableConfig tableOptions, int[] primaryKeyIndexes) {
+ if (!tableOptions.getOptional(KEY_FIELDS).isPresent()) {
+ throw new ValidationException(
+ "Upsert mode requires key.fields set to the primary key fields, should be set");
+ }
+
+ if (tableOptions.getOptional(KEY_FIELDS).get().size() == 0
+ || primaryKeyIndexes.length == 0) {
+ throw new ValidationException(
+ "'upsert-pulsar' require to define a PRIMARY KEY constraint. "
Review Comment:
Maybe I can include it in this PR. Let me have a try.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304189895
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
Review Comment:
Do you mean enrich the document? It's not reasonable to add runtime check but let the Flink Table kernel throws exceptions.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304185559
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
Review Comment:
No. Support `topic-pattern` can be an advanced feature. I don't think it's a blocker of this patch.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-XXXXX] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1272051816
##########
flink-connector-pulsar/src/test/resources/log4j2-test.properties:
##########
@@ -18,7 +18,7 @@
# Set root logger level to OFF to not flood build logs
# set manually to INFO for debugging purposes
-rootLogger.level=OFF
+rootLogger.level=INFO
Review Comment:
To-be reverted.
--
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-connector-pulsar] tisonkun commented on pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1655131941
This patch itself is ready for review and technically mergable. Feel free to leave a comment.
cc @nlu90.
--
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-connector-pulsar] tisonkun commented on pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1700296791
@leonardBang Thank you! Merged.
> sql jar and task for docs can start as well next
Yep. Let me create the related JIRA tickets..
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304164089
##########
pom.xml:
##########
@@ -106,6 +110,7 @@ under the License.
<scope>provided</scope>
</dependency>
+
Review Comment:
Reverted.
--
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-connector-pulsar] leonardBang commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1303009152
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableFactory.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.pulsar.common.config.PulsarOptions;
+import org.apache.flink.connector.pulsar.sink.PulsarSinkOptions;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.table.sink.PulsarTableSerializationSchemaFactory;
+import org.apache.flink.connector.pulsar.table.sink.PulsarTableSink;
+import org.apache.flink.connector.pulsar.table.source.PulsarTableDeserializationSchemaFactory;
+import org.apache.flink.connector.pulsar.table.source.PulsarTableSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.createKeyFormatProjection;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.createValueFormatProjection;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getKeyDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getKeyEncodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getMessageDelayMillis;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getPulsarProperties;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getStartCursor;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getStopCursor;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getSubscriptionType;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getTopicListFromOptions;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getTopicRouter;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getTopicRoutingMode;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueEncodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.ADMIN_URL;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.EXPLICIT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FIELDS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SERVICE_URL;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_CUSTOM_TOPIC_ROUTER;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_MESSAGE_DELAY_INTERVAL;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_TOPIC_ROUTING_MODE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AFTER_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.TOPICS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.VALUE_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableValidationUtils.validatePrimaryKeyConstraints;
+import static org.apache.flink.connector.pulsar.table.PulsarTableValidationUtils.validateTableSinkOptions;
+import static org.apache.flink.connector.pulsar.table.PulsarTableValidationUtils.validateTableSourceOptions;
+import static org.apache.flink.table.factories.FactoryUtil.SINK_PARALLELISM;
+import static org.apache.pulsar.shade.org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+
+/**
+ * Factory for creating {@link DynamicTableSource} and {@link DynamicTableSink}.
+ *
+ * <p>The main role of this class is to retrieve config options and validate options from config and
+ * the table schema. It also sets default values if a config option is not present.
+ */
+public class PulsarTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ public static final String IDENTIFIER = "pulsar";
+
+ public static final String DEFAULT_SUBSCRIPTION_NAME_PREFIX = "flink-sql-connector-pulsar-";
+
+ public static final boolean UPSERT_DISABLED = false;
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ // Format options should be retrieved before validation.
+ final DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat =
+ getKeyDecodingFormat(helper);
+ final DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat =
+ getValueDecodingFormat(helper);
+ ReadableConfig tableOptions = helper.getOptions();
+
+ // Validate configs are not conflict; each options is consumed; no unwanted configs
+ // PulsarOptions, PulsarSourceOptions and PulsarSinkOptions is not part of the validation.
+ helper.validateExcept(
+ PulsarOptions.CLIENT_CONFIG_PREFIX,
+ PulsarOptions.ADMIN_CONFIG_PREFIX,
+ PulsarSourceOptions.SOURCE_CONFIG_PREFIX,
+ PulsarSourceOptions.CONSUMER_CONFIG_PREFIX,
+ PulsarSinkOptions.PRODUCER_CONFIG_PREFIX,
+ PulsarSinkOptions.SINK_CONFIG_PREFIX);
+
+ validatePrimaryKeyConstraints(
+ context.getObjectIdentifier(), context.getPrimaryKeyIndexes(), helper);
+
+ validateTableSourceOptions(tableOptions);
+
+ // Retrieve configs
+ final List<String> topics = getTopicListFromOptions(tableOptions);
+ final StartCursor startCursor = getStartCursor(tableOptions);
+ final StopCursor stopCursor = getStopCursor(tableOptions);
+ final SubscriptionType subscriptionType = getSubscriptionType(tableOptions);
+
+ // Forward source configs
+ final Properties properties = getPulsarProperties(tableOptions);
+ properties.setProperty(PULSAR_ADMIN_URL.key(), tableOptions.get(ADMIN_URL));
+ properties.setProperty(PULSAR_SERVICE_URL.key(), tableOptions.get(SERVICE_URL));
+ // Set random subscriptionName if not provided
+ properties.setProperty(
+ PULSAR_SUBSCRIPTION_NAME.key(),
+ tableOptions
+ .getOptional(SOURCE_SUBSCRIPTION_NAME)
+ .orElse(DEFAULT_SUBSCRIPTION_NAME_PREFIX + randomAlphabetic(5)));
+ // Retrieve physical fields (not including computed or metadata fields),
+ // and projections and create a schema factory based on such information.
+ final DataType physicalDataType = context.getPhysicalRowDataType();
+
+ final int[] valueProjection = createValueFormatProjection(tableOptions, physicalDataType);
+ final int[] keyProjection = createKeyFormatProjection(tableOptions, physicalDataType);
+
+ final PulsarTableDeserializationSchemaFactory deserializationSchemaFactory =
+ new PulsarTableDeserializationSchemaFactory(
+ physicalDataType,
+ keyDecodingFormat,
+ keyProjection,
+ valueDecodingFormat,
+ valueProjection,
+ UPSERT_DISABLED);
+
+ // Set default values for configuration not exposed to user.
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormatForMetadataPushdown =
+ valueDecodingFormat;
+ final ChangelogMode changelogMode = decodingFormatForMetadataPushdown.getChangelogMode();
+
+ return new PulsarTableSource(
+ deserializationSchemaFactory,
+ decodingFormatForMetadataPushdown,
+ changelogMode,
+ topics,
+ properties,
+ startCursor,
+ stopCursor,
+ subscriptionType);
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ // Format options should be retrieved before validation.
+ final EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat =
+ getKeyEncodingFormat(helper);
+ final EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat =
+ getValueEncodingFormat(helper);
+ ReadableConfig tableOptions = helper.getOptions();
+
+ // Validate configs are not conflict; each options is consumed; no unwanted configs
+ // PulsarOptions, PulsarSourceOptions and PulsarSinkOptions is not part of the validation.
+ helper.validateExcept(
+ PulsarOptions.CLIENT_CONFIG_PREFIX,
+ PulsarOptions.ADMIN_CONFIG_PREFIX,
+ PulsarSourceOptions.SOURCE_CONFIG_PREFIX,
+ PulsarSourceOptions.CONSUMER_CONFIG_PREFIX,
+ PulsarSinkOptions.PRODUCER_CONFIG_PREFIX,
+ PulsarSinkOptions.SINK_CONFIG_PREFIX);
+
+ validatePrimaryKeyConstraints(
+ context.getObjectIdentifier(), context.getPrimaryKeyIndexes(), helper);
+
+ validateTableSinkOptions(tableOptions);
+
+ // Retrieve configs
+ final TopicRouter<RowData> topicRouter =
+ getTopicRouter(tableOptions, context.getClassLoader());
+ final TopicRoutingMode topicRoutingMode = getTopicRoutingMode(tableOptions);
+ final long messageDelayMillis = getMessageDelayMillis(tableOptions);
+
+ final List<String> topics = getTopicListFromOptions(tableOptions);
+
+ // Forward sink configs
+ final Properties properties = getPulsarProperties(tableOptions);
+ properties.setProperty(PULSAR_ADMIN_URL.key(), tableOptions.get(ADMIN_URL));
+ properties.setProperty(PULSAR_SERVICE_URL.key(), tableOptions.get(SERVICE_URL));
+
+ // Retrieve physical DataType (not including computed or metadata fields)
+ final DataType physicalDataType = context.getPhysicalRowDataType();
+ final int[] keyProjection = createKeyFormatProjection(tableOptions, physicalDataType);
+ final int[] valueProjection = createValueFormatProjection(tableOptions, physicalDataType);
+
+ final PulsarTableSerializationSchemaFactory serializationSchemaFactory =
+ new PulsarTableSerializationSchemaFactory(
+ physicalDataType,
+ keyEncodingFormat,
+ keyProjection,
+ valueEncodingFormat,
+ valueProjection,
+ UPSERT_DISABLED);
+
+ // Set default values for configuration not exposed to user.
+ final DeliveryGuarantee deliveryGuarantee = DeliveryGuarantee.AT_LEAST_ONCE;
+ final ChangelogMode changelogMode = valueEncodingFormat.getChangelogMode();
+
+ return new PulsarTableSink(
+ serializationSchemaFactory,
+ changelogMode,
+ topics,
+ properties,
+ deliveryGuarantee,
+ topicRouter,
+ topicRoutingMode,
+ messageDelayMillis);
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Stream.of(TOPICS, ADMIN_URL, SERVICE_URL).collect(Collectors.toSet());
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Stream.of(
+ FactoryUtil.FORMAT,
+ VALUE_FORMAT,
+ SOURCE_SUBSCRIPTION_NAME,
+ SOURCE_SUBSCRIPTION_TYPE,
+ SOURCE_START_FROM_MESSAGE_ID,
+ SOURCE_START_FROM_PUBLISH_TIME,
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME,
+ SINK_CUSTOM_TOPIC_ROUTER,
+ SINK_TOPIC_ROUTING_MODE,
+ SINK_MESSAGE_DELAY_INTERVAL,
+ SINK_PARALLELISM,
+ KEY_FORMAT,
+ KEY_FIELDS,
+ EXPLICIT)
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * Format and Delivery guarantee related options are not forward options.
+ *
+ * @return
Review Comment:
minor: remove or add some words
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
Review Comment:
These topics should have same schema for data as Flink Table need a fixed schema, could you add the limitation?
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
Review Comment:
I'm a little confused about the two configurations from users' perspective, why `subscription-type` has a default value and the later one does not?
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSink.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * Licensed 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.sink;
+
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.pulsar.sink.PulsarSink;
+import org.apache.flink.connector.pulsar.sink.PulsarSinkBuilder;
+import org.apache.flink.connector.pulsar.sink.writer.delayer.FixedMessageDelayer;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsWritingMetadata;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.stream.Stream;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Pulsar SQL Connector sink. It supports {@link SupportsWritingMetadata}. */
+public class PulsarTableSink implements DynamicTableSink, SupportsWritingMetadata {
+
+ private final PulsarTableSerializationSchemaFactory serializationSchemaFactory;
+
+ private final ChangelogMode changelogMode;
+
+ private final List<String> topics;
+
+ private final Properties properties;
+
+ private final DeliveryGuarantee deliveryGuarantee;
+
+ @Nullable private final TopicRouter<RowData> topicRouter;
+
+ private final TopicRoutingMode topicRoutingMode;
+
+ private final long messageDelayMillis;
+
+ public PulsarTableSink(
+ PulsarTableSerializationSchemaFactory serializationSchemaFactory,
+ ChangelogMode changelogMode,
+ List<String> topics,
+ Properties properties,
+ DeliveryGuarantee deliveryGuarantee,
+ @Nullable TopicRouter<RowData> topicRouter,
+ TopicRoutingMode topicRoutingMode,
+ long messageDelayMillis) {
+ this.serializationSchemaFactory = checkNotNull(serializationSchemaFactory);
+ this.changelogMode = checkNotNull(changelogMode);
+ this.topics = checkNotNull(topics);
+ // Mutable attributes
+ this.properties = checkNotNull(properties);
+ this.deliveryGuarantee = checkNotNull(deliveryGuarantee);
+ this.topicRouter = topicRouter;
+ this.topicRoutingMode = checkNotNull(topicRoutingMode);
+ this.messageDelayMillis = messageDelayMillis;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return this.changelogMode;
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+
+ final PulsarSerializationSchema<RowData> pulsarSerializationSchema =
+ serializationSchemaFactory.createPulsarSerializationSchema(context);
+
+ final PulsarSinkBuilder<RowData> pulsarSinkBuilder =
+ PulsarSink.builder()
+ .setSerializationSchema(pulsarSerializationSchema)
+ .setProperties(properties)
+ .setDeliveryGuarantee(deliveryGuarantee)
+ .setTopics(topics)
+ .setTopicRoutingMode(topicRoutingMode)
+ .delaySendingMessage(new FixedMessageDelayer<>(messageDelayMillis));
Review Comment:
Here looks like we set a different default value.
In table: default 0 will map to FixedMessageDelayer
In DS：default -1 map to never() FixedMessageDelayer
Duration of 0 is meaningless for most cases as publish/schedule with a delay 0 mills is meaningless, I'd like to use -1 as default value.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableValidationUtils.java:
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.RowKind;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.shade.com.google.common.collect.Sets;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FIELDS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_CUSTOM_TOPIC_ROUTER;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_TOPIC_ROUTING_MODE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AFTER_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.TOPICS;
+import static org.apache.pulsar.common.naming.TopicName.isValid;
+
+/** Util class for source and sink validation rules. */
+public class PulsarTableValidationUtils {
+
+ private PulsarTableValidationUtils() {}
+
+ public static void validatePrimaryKeyConstraints(
+ ObjectIdentifier tableName,
+ int[] primaryKeyIndexes,
+ FactoryUtil.TableFactoryHelper helper) {
+ final DecodingFormat<DeserializationSchema<RowData>> format =
+ getValueDecodingFormat(helper);
+ if (primaryKeyIndexes.length > 0
+ && format.getChangelogMode().containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ String.format(
+ "The Pulsar table '%s' with '%s' format doesn't support defining PRIMARY KEY constraint"
+ + " on the table, because it can't guarantee the semantic of primary key.",
+ tableName.asSummaryString(), format));
+ }
+ }
+
+ public static void validateTableSourceOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateStartCursorConfigs(tableOptions);
+ validateStopCursorConfigs(tableOptions);
+ validateSubscriptionTypeConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ }
+
+ public static void validateTableSinkOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ validateSinkRoutingConfigs(tableOptions);
+ }
+
+ protected static void validateTopicsConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.get(TOPICS).isEmpty()) {
+ throw new ValidationException("The topics list should not be empty.");
+ }
+
+ for (String topic : tableOptions.get(TOPICS)) {
+ if (!isValid(topic)) {
+ throw new ValidationException(
+ String.format("The topics name %s is not a valid topic name.", topic));
+ }
+ }
+ }
+
+ protected static void validateStartCursorConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SOURCE_START_FROM_MESSAGE_ID).isPresent()
+ && tableOptions.getOptional(SOURCE_START_FROM_PUBLISH_TIME).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SOURCE_START_FROM_MESSAGE_ID, SOURCE_START_FROM_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateStopCursorConfigs(ReadableConfig tableOptions) {
+ Set<ConfigOption<?>> conflictConfigOptions =
+ Sets.newHashSet(
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME);
+
+ long configsNums =
+ conflictConfigOptions.stream()
+ .map(tableOptions::getOptional)
+ .filter(Optional::isPresent)
+ .count();
+
+ if (configsNums > 1) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s, %s and %s can be specified. Detected more than 1 of them",
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateSubscriptionTypeConfigs(ReadableConfig tableOptions) {
+ SubscriptionType subscriptionType = tableOptions.get(SOURCE_SUBSCRIPTION_TYPE);
+
+ if (subscriptionType == SubscriptionType.Failover) {
+ throw new ValidationException(
+ String.format(
+ "%s SubscriptionType is not supported. ", SubscriptionType.Failover));
+ }
+ }
+
+ protected static void validateKeyFormatConfigs(ReadableConfig tableOptions) {
+ final Optional<String> optionalKeyFormat = tableOptions.getOptional(KEY_FORMAT);
+ final Optional<List<String>> optionalKeyFields = tableOptions.getOptional(KEY_FIELDS);
+ if (!optionalKeyFormat.isPresent() && optionalKeyFields.isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "The option '%s' can only be declared if a key format is defined using '%s'.",
+ KEY_FIELDS.key(), KEY_FORMAT.key()));
+ } else if (optionalKeyFormat.isPresent()
+ && (!optionalKeyFields.isPresent() || optionalKeyFields.get().size() == 0)) {
+ throw new ValidationException(
+ String.format(
+ "A key format '%s' requires the declaration of one or more of key fields using '%s'.",
+ KEY_FORMAT.key(), KEY_FIELDS.key()));
+ }
+ }
+
+ protected static void validateSinkRoutingConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).orElse(TopicRoutingMode.ROUND_ROBIN)
+ == TopicRoutingMode.CUSTOM) {
+ throw new ValidationException(
+ String.format(
+ "Only %s and %s can be used. For %s, please use sink.custom-topic-router for"
+ + "custom topic router and do not set this config.",
+ TopicRoutingMode.ROUND_ROBIN,
+ TopicRoutingMode.MESSAGE_KEY_HASH,
+ TopicRoutingMode.CUSTOM));
+ }
+ if (tableOptions.getOptional(SINK_CUSTOM_TOPIC_ROUTER).isPresent()
+ && tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SINK_CUSTOM_TOPIC_ROUTER, SINK_TOPIC_ROUTING_MODE));
+ }
+ }
+
+ protected static void validateUpsertModeKeyConstraints(
+ ReadableConfig tableOptions, int[] primaryKeyIndexes) {
+ if (!tableOptions.getOptional(KEY_FIELDS).isPresent()) {
+ throw new ValidationException(
+ "Upsert mode requires key.fields set to the primary key fields, should be set");
+ }
+
+ if (tableOptions.getOptional(KEY_FIELDS).get().size() == 0
+ || primaryKeyIndexes.length == 0) {
+ throw new ValidationException(
+ "'upsert-pulsar' require to define a PRIMARY KEY constraint. "
Review Comment:
`upsert-pulsar` did expose to users yet, right? I tend to add a upsert-pulsar table connector, it should be different with current pulsar table connector
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarRowDataConverter.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.DeserializationException;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.Collector;
+
+import org.apache.pulsar.client.api.Message;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Contains the projection information needed to map a Pulsar message to proper key fields, value
+ * fields and metadata fields.
+ */
+public class PulsarRowDataConverter implements Serializable {
+ private static final long serialVersionUID = -3399264407634977459L;
Review Comment:
1L
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
+ + "source. Use \"never\", \"latest\" or pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\"");
+
+ public static final ConfigOption<String> SOURCE_STOP_AFTER_MESSAGE_ID =
+ ConfigOptions.key("source.stop.after-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop position but include the "
+ + "given message in the consuming result for the unbounded sql "
+ + "source. Pass in a message id "
Review Comment:
bounded ？
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
Review Comment:
unbounded？
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchema.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.api.common.functions.util.ListCollector;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+
+import org.apache.pulsar.client.api.Message;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A specific {@link PulsarDeserializationSchema} for {@link PulsarTableSource}.
+ *
+ * <p>Both Flink's key decoding format and value decoding format are wrapped in this class. It is
+ * responsible for getting metadata fields from a physical pulsar message body, and the final
+ * projection mapping from Pulsar message fields to Flink row.
+ *
+ * <p>After retrieving key and value bytes and convert them into a list of {@link RowData}, it then
+ * delegates metadata appending, key and value {@link RowData} combining to a {@link
+ * PulsarRowDataConverter} instance.
+ */
+public class PulsarTableDeserializationSchema implements PulsarDeserializationSchema<RowData> {
+ private static final long serialVersionUID = -3298784447432136216L;
Review Comment:
1L
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
+ + "source. Use \"never\", \"latest\" or pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\"");
+
+ public static final ConfigOption<String> SOURCE_STOP_AFTER_MESSAGE_ID =
+ ConfigOptions.key("source.stop.after-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop position but include the "
+ + "given message in the consuming result for the unbounded sql "
+ + "source. Pass in a message id "
Review Comment:
bounded ？
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchema.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.api.common.functions.util.ListCollector;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+
+import org.apache.pulsar.client.api.Message;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A specific {@link PulsarDeserializationSchema} for {@link PulsarTableSource}.
+ *
+ * <p>Both Flink's key decoding format and value decoding format are wrapped in this class. It is
+ * responsible for getting metadata fields from a physical pulsar message body, and the final
+ * projection mapping from Pulsar message fields to Flink row.
+ *
+ * <p>After retrieving key and value bytes and convert them into a list of {@link RowData}, it then
+ * delegates metadata appending, key and value {@link RowData} combining to a {@link
+ * PulsarRowDataConverter} instance.
+ */
+public class PulsarTableDeserializationSchema implements PulsarDeserializationSchema<RowData> {
+ private static final long serialVersionUID = -3298784447432136216L;
+
+ private final TypeInformation<RowData> producedTypeInfo;
+
+ @Nullable private final DeserializationSchema<RowData> keyDeserialization;
+
+ private final DeserializationSchema<RowData> valueDeserialization;
+
+ private final PulsarRowDataConverter rowDataConverter;
+
+ private final boolean upsertMode;
+
+ public PulsarTableDeserializationSchema(
+ @Nullable DeserializationSchema<RowData> keyDeserialization,
+ DeserializationSchema<RowData> valueDeserialization,
+ TypeInformation<RowData> producedTypeInfo,
+ PulsarRowDataConverter rowDataConverter,
+ boolean upsertMode) {
+ this.keyDeserialization = keyDeserialization;
+ this.valueDeserialization = checkNotNull(valueDeserialization);
+ this.rowDataConverter = checkNotNull(rowDataConverter);
+ this.producedTypeInfo = checkNotNull(producedTypeInfo);
+ this.upsertMode = upsertMode;
+ }
+
+ @Override
+ public void open(PulsarInitializationContext context, SourceConfiguration configuration)
+ throws Exception {
+ if (keyDeserialization != null) {
+ keyDeserialization.open(context);
+ }
+ valueDeserialization.open(context);
+ }
+
+ @Override
+ public void deserialize(Message<byte[]> message, Collector<RowData> collector)
+ throws IOException {
+
+ // Get the key row data
+ List<RowData> keyRowData = new ArrayList<>();
+ if (keyDeserialization != null) {
+ keyDeserialization.deserialize(message.getKeyBytes(), new ListCollector<>(keyRowData));
+ }
+
+ // Get the value row data
+ List<RowData> valueRowData = new ArrayList<>();
+
+ if (upsertMode && message.getData().length == 0) {
+ checkNotNull(keyDeserialization, "upsert mode must specify a key format");
Review Comment:
We can move the check to compile phase to check once instead of checking for each tombstone message
##########
pom.xml:
##########
@@ -106,6 +110,7 @@ under the License.
<scope>provided</scope>
</dependency>
+
Review Comment:
useless line
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304191186
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
+ + "source. Use \"never\", \"latest\" or pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\"");
+
+ public static final ConfigOption<String> SOURCE_STOP_AFTER_MESSAGE_ID =
+ ConfigOptions.key("source.stop.after-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop position but include the "
+ + "given message in the consuming result for the unbounded sql "
+ + "source. Pass in a message id "
Review Comment:
ditto https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304172205
--
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-connector-pulsar] tisonkun commented on pull request #56: [FLINK-XXXXX] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1647545819
cc @syhily @cbornet I'll appreciate it if you can help with review and suggest on code or features.
I'm pending to add you two and @imaffe to the Co-authored-by columns.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-XXXXX] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1271787129
##########
flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableITCase.java:
##########
@@ -0,0 +1,678 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.connector.pulsar.table.testutils.TestingUser;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.test.util.SuccessException;
+import org.apache.flink.types.Row;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Stream;
+
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+import static org.apache.flink.connector.pulsar.table.testutils.PulsarTableTestUtils.collectRows;
+import static org.apache.flink.connector.pulsar.table.testutils.TestingUser.createRandomUser;
+import static org.apache.flink.util.CollectionUtil.entry;
+import static org.apache.flink.util.CollectionUtil.map;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+
+/**
+ * IT cases for the Pulsar table source and sink. It aims to verify runtime behaviour and certain
+ * use cases are correct and can produce/consume the desired records as user specifies.
+ */
+@ExtendWith(MiniClusterExtension.class)
+public class PulsarTableITCase extends PulsarTableTestBase {
+
+ protected static final String JSON_FORMAT = "json";
+ protected static final String AVRO_FORMAT = "avro";
+ protected static final String CSV_FORMAT = "csv";
+ protected static final String RAW_FORMAT = "raw";
+
+ @ParameterizedTest
+ @ValueSource(strings = {JSON_FORMAT, AVRO_FORMAT, CSV_FORMAT})
+ void pulsarSourceSink(String format) throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "test_topic_" + format + randomAlphanumeric(3);
+ createTestTopic(topic, 1);
+
+ // ---------- Produce an event time stream into Pulsar -------------------
+ String randomTableName = randomAlphabetic(5);
+ final String createTable =
+ String.format(
+ "create table %s (\n"
+ + " `computed-price` as price + 1.0,\n"
+ + " price decimal(38, 18),\n"
+ + " currency string,\n"
+ + " log_date date,\n"
+ + " log_time time(3),\n"
+ + " log_ts timestamp(3),\n"
+ + " ts as log_ts + INTERVAL '1' SECOND,\n"
+ + " watermark for ts as ts\n"
+ + ") with (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topics' = '%s',\n"
+ + " 'service-url' = '%s',\n"
+ + " 'admin-url' = '%s',\n"
+ + " 'format' = '%s'\n"
+ + ")",
+ randomTableName,
+ PulsarTableFactory.IDENTIFIER,
+ topic,
+ pulsar.operator().serviceUrl(),
+ pulsar.operator().adminUrl(),
+ format);
+
+ tableEnv.executeSql(createTable);
+
+ String initialValues =
+ String.format(
+ "INSERT INTO %s\n"
+ + "SELECT CAST(price AS DECIMAL(10, 2)), currency, "
+ + " CAST(d AS DATE), CAST(t AS TIME(0)), CAST(ts AS TIMESTAMP(3))\n"
+ + "FROM (VALUES (2.02,'Euro','2019-12-12', '00:00:01', '2019-12-12 00:00:01.001001'), \n"
+ + " (1.11,'US Dollar','2019-12-12', '00:00:02', '2019-12-12 00:00:02.002001'), \n"
+ + " (50,'Yen','2019-12-12', '00:00:03', '2019-12-12 00:00:03.004001'), \n"
+ + " (3.1,'Euro','2019-12-12', '00:00:04', '2019-12-12 00:00:04.005001'), \n"
+ + " (5.33,'US Dollar','2019-12-12', '00:00:05', '2019-12-12 00:00:05.006001'), \n"
+ + " (0,'DUMMY','2019-12-12', '00:00:10', '2019-12-12 00:00:10'))\n"
+ + " AS orders (price, currency, d, t, ts)",
+ randomTableName);
+ tableEnv.executeSql(initialValues).await();
+
+ String query =
+ String.format(
+ "SELECT\n"
+ + " CAST(TUMBLE_END(ts, INTERVAL '5' SECOND) AS VARCHAR),\n"
+ + " CAST(MAX(log_date) AS VARCHAR),\n"
+ + " CAST(MAX(log_time) AS VARCHAR),\n"
+ + " CAST(MAX(ts) AS VARCHAR),\n"
+ + " COUNT(*),\n"
+ + " CAST(MAX(price) AS DECIMAL(10, 2))\n"
+ + "FROM %s\n"
+ + "GROUP BY TUMBLE(ts, INTERVAL '5' SECOND)",
+ randomTableName);
+
+ DataStream<Row> result = tableEnv.toDataStream(tableEnv.sqlQuery(query));
+ TestingSinkFunction sink = new TestingSinkFunction(2);
+ result.addSink(sink).setParallelism(1);
+
+ try {
+ env.execute("Job_2");
+ } catch (Throwable e) {
+ if (!isCausedByJobFinished(e)) {
+ // re-throw
+ throw e;
+ }
+ }
+
+ List<String> expected =
+ Arrays.asList(
+ "+I[2019-12-12 00:00:05.000, 2019-12-12, 00:00:03, 2019-12-12 00:00:04.004, 3, 50.00]",
+ "+I[2019-12-12 00:00:10.000, 2019-12-12, 00:00:05, 2019-12-12 00:00:06.006, 2, 5.33]");
+
+ assertThat(TestingSinkFunction.rows).isEqualTo(expected);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {JSON_FORMAT, AVRO_FORMAT, CSV_FORMAT})
+ void pulsarSourceSinkWithKeyAndPartialValue(String format) throws Exception {
Review Comment:
This test still doesn't pass. But maybe we don't support this use case at all.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304184867
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
Review Comment:
... and we actually fallback to a default subscription name in code:
```java
.getOptional(SOURCE_SUBSCRIPTION_NAME)
.orElse(DEFAULT_SUBSCRIPTION_NAME_PREFIX + randomAlphabetic(5)));
```
--
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-connector-pulsar] leonardBang commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1303026417
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
Review Comment:
we use `topic` and `topic-pattern ` in kafka connector, I wonder could we unified this concept?
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
+ + "source. Use \"never\", \"latest\" or pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\"");
+
+ public static final ConfigOption<String> SOURCE_STOP_AFTER_MESSAGE_ID =
+ ConfigOptions.key("source.stop.after-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop position but include the "
+ + "given message in the consuming result for the unbounded sql "
+ + "source. Pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\". ");
+
+ public static final ConfigOption<Long> SOURCE_STOP_AT_PUBLISH_TIME =
+ ConfigOptions.key("source.stop.at-publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional publish timestamp used to specify a stop cursor"
+ + " for the unbounded sql source.");
Review Comment:
bounded
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchemaFactory.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.table.connector.Projection;
+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.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Contains key, value projection and format information, and use such information to create a
+ * {@link PulsarTableDeserializationSchema} instance used by runtime {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSource} instance.
+ *
+ * <p>A Flink row fields has a strict order: Physical Fields (Key + value) + Format Metadata Fields
+ * Connector Metadata Fields. Physical Fields are fields come directly from Pulsar message body;
+ * Format Metadata Fields are from the extra information from the decoding format. Connector
+ * metadata fields are the ones most Pulsar messages have, such as publish time, message size and
+ * producer name.
+ *
+ * <p>In general, Physical fields + Format Metadata fields are contained in the RowData decoded
+ * using valueDecodingFormat. Only Connector Metadata fields needs to be appended to the decoded
+ * RowData. The tricky part is to put format metadata and connector metadata in the right location.
+ * This requires an explicit adjustment process.
+ *
+ * <p>For example, suppose Physical Fields (Key + value) + Format Metadata Fields + Connector
+ * Metadata Fields. has arity of 11, key projection is [0, 6], and physical value projection is [1,
+ * 2, 3, 4, 5], Then after the adjustment, key projection should be [0, 6], physical value
+ * projection should be [1, 2, 3, 4, 5] and format metadata projection should be [7], connector
+ * metadata projection should be [8, 9, 10].
+ */
+public class PulsarTableDeserializationSchemaFactory implements Serializable {
+
+ private static final long serialVersionUID = 6091562041940740434L;
Review Comment:
1L
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
Review Comment:
The `earliest` and `latest` worth a public configuration for users from my side, these startup modes are similar to specific-offset in kafka connector. I don't think the configuration in Kafka connector is good one, I just want to align them for decreasing the users' understanding cost.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarWritableMetadata.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.sink;
+
+import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessageBuilder;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import org.bouncycastle.util.Arrays;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Stream;
+
+/** A class used to manage WritableMetadata for Pulsar SQL sink connector. */
+public class PulsarWritableMetadata implements Serializable {
+
+ private static final long serialVersionUID = 8117156158379846715L;
Review Comment:
1L
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSerializationSchema.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * Licensed 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
+import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
+import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessage;
+import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessageBuilder;
+import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
+
+import org.apache.pulsar.client.api.Schema;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@link PulsarSerializationSchema} implementation for Pulsar SQL sink connector. It is
+ * responsible for retrieving fields from Flink row and serialize into Pulsar message key or body,
+ * and set necessary metadata fields as required.
+ */
+public class PulsarTableSerializationSchema implements PulsarSerializationSchema<RowData> {
+
+ private static final long serialVersionUID = 7314442107082067836L;
Review Comment:
IIRC, Flink uses serialVersionUID from `1L` by default?
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
Review Comment:
unbounded？
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSink.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * Licensed 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.sink;
+
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.pulsar.sink.PulsarSink;
+import org.apache.flink.connector.pulsar.sink.PulsarSinkBuilder;
+import org.apache.flink.connector.pulsar.sink.writer.delayer.FixedMessageDelayer;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsWritingMetadata;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.stream.Stream;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Pulsar SQL Connector sink. It supports {@link SupportsWritingMetadata}. */
+public class PulsarTableSink implements DynamicTableSink, SupportsWritingMetadata {
+
+ private final PulsarTableSerializationSchemaFactory serializationSchemaFactory;
+
+ private final ChangelogMode changelogMode;
+
+ private final List<String> topics;
+
+ private final Properties properties;
+
+ private final DeliveryGuarantee deliveryGuarantee;
+
+ @Nullable private final TopicRouter<RowData> topicRouter;
+
+ private final TopicRoutingMode topicRoutingMode;
+
+ private final long messageDelayMillis;
+
+ public PulsarTableSink(
+ PulsarTableSerializationSchemaFactory serializationSchemaFactory,
+ ChangelogMode changelogMode,
+ List<String> topics,
+ Properties properties,
+ DeliveryGuarantee deliveryGuarantee,
+ @Nullable TopicRouter<RowData> topicRouter,
+ TopicRoutingMode topicRoutingMode,
+ long messageDelayMillis) {
+ this.serializationSchemaFactory = checkNotNull(serializationSchemaFactory);
+ this.changelogMode = checkNotNull(changelogMode);
+ this.topics = checkNotNull(topics);
+ // Mutable attributes
+ this.properties = checkNotNull(properties);
+ this.deliveryGuarantee = checkNotNull(deliveryGuarantee);
+ this.topicRouter = topicRouter;
+ this.topicRoutingMode = checkNotNull(topicRoutingMode);
+ this.messageDelayMillis = messageDelayMillis;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return this.changelogMode;
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+
+ final PulsarSerializationSchema<RowData> pulsarSerializationSchema =
+ serializationSchemaFactory.createPulsarSerializationSchema(context);
+
+ final PulsarSinkBuilder<RowData> pulsarSinkBuilder =
+ PulsarSink.builder()
+ .setSerializationSchema(pulsarSerializationSchema)
+ .setProperties(properties)
+ .setDeliveryGuarantee(deliveryGuarantee)
+ .setTopics(topics)
+ .setTopicRoutingMode(topicRoutingMode)
+ .delaySendingMessage(new FixedMessageDelayer<>(messageDelayMillis));
Review Comment:
Here looks like we set a different default value.
In table: default 0 will map to FixedMessageDelayer
In DS：default -1 map to never() FixedMessageDelayer
Duration of 0 is meaningless for most cases as publish/schedule with a delay 0 mills is meaningless, I'd like to use -1 as default value.
--
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-connector-pulsar] tisonkun merged pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun merged PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304194301
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchema.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.api.common.functions.util.ListCollector;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+
+import org.apache.pulsar.client.api.Message;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A specific {@link PulsarDeserializationSchema} for {@link PulsarTableSource}.
+ *
+ * <p>Both Flink's key decoding format and value decoding format are wrapped in this class. It is
+ * responsible for getting metadata fields from a physical pulsar message body, and the final
+ * projection mapping from Pulsar message fields to Flink row.
+ *
+ * <p>After retrieving key and value bytes and convert them into a list of {@link RowData}, it then
+ * delegates metadata appending, key and value {@link RowData} combining to a {@link
+ * PulsarRowDataConverter} instance.
+ */
+public class PulsarTableDeserializationSchema implements PulsarDeserializationSchema<RowData> {
+ private static final long serialVersionUID = -3298784447432136216L;
+
+ private final TypeInformation<RowData> producedTypeInfo;
+
+ @Nullable private final DeserializationSchema<RowData> keyDeserialization;
+
+ private final DeserializationSchema<RowData> valueDeserialization;
+
+ private final PulsarRowDataConverter rowDataConverter;
+
+ private final boolean upsertMode;
+
+ public PulsarTableDeserializationSchema(
+ @Nullable DeserializationSchema<RowData> keyDeserialization,
+ DeserializationSchema<RowData> valueDeserialization,
+ TypeInformation<RowData> producedTypeInfo,
+ PulsarRowDataConverter rowDataConverter,
+ boolean upsertMode) {
+ this.keyDeserialization = keyDeserialization;
+ this.valueDeserialization = checkNotNull(valueDeserialization);
+ this.rowDataConverter = checkNotNull(rowDataConverter);
+ this.producedTypeInfo = checkNotNull(producedTypeInfo);
+ this.upsertMode = upsertMode;
+ }
+
+ @Override
+ public void open(PulsarInitializationContext context, SourceConfiguration configuration)
+ throws Exception {
+ if (keyDeserialization != null) {
+ keyDeserialization.open(context);
+ }
+ valueDeserialization.open(context);
+ }
+
+ @Override
+ public void deserialize(Message<byte[]> message, Collector<RowData> collector)
+ throws IOException {
+
+ // Get the key row data
+ List<RowData> keyRowData = new ArrayList<>();
+ if (keyDeserialization != null) {
+ keyDeserialization.deserialize(message.getKeyBytes(), new ListCollector<>(keyRowData));
+ }
+
+ // Get the value row data
+ List<RowData> valueRowData = new ArrayList<>();
+
+ if (upsertMode && message.getData().length == 0) {
+ checkNotNull(keyDeserialization, "upsert mode must specify a key format");
Review Comment:
Updated.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304198565
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSink.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * Licensed 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.sink;
+
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.pulsar.sink.PulsarSink;
+import org.apache.flink.connector.pulsar.sink.PulsarSinkBuilder;
+import org.apache.flink.connector.pulsar.sink.writer.delayer.FixedMessageDelayer;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsWritingMetadata;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.stream.Stream;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Pulsar SQL Connector sink. It supports {@link SupportsWritingMetadata}. */
+public class PulsarTableSink implements DynamicTableSink, SupportsWritingMetadata {
+
+ private final PulsarTableSerializationSchemaFactory serializationSchemaFactory;
+
+ private final ChangelogMode changelogMode;
+
+ private final List<String> topics;
+
+ private final Properties properties;
+
+ private final DeliveryGuarantee deliveryGuarantee;
+
+ @Nullable private final TopicRouter<RowData> topicRouter;
+
+ private final TopicRoutingMode topicRoutingMode;
+
+ private final long messageDelayMillis;
+
+ public PulsarTableSink(
+ PulsarTableSerializationSchemaFactory serializationSchemaFactory,
+ ChangelogMode changelogMode,
+ List<String> topics,
+ Properties properties,
+ DeliveryGuarantee deliveryGuarantee,
+ @Nullable TopicRouter<RowData> topicRouter,
+ TopicRoutingMode topicRoutingMode,
+ long messageDelayMillis) {
+ this.serializationSchemaFactory = checkNotNull(serializationSchemaFactory);
+ this.changelogMode = checkNotNull(changelogMode);
+ this.topics = checkNotNull(topics);
+ // Mutable attributes
+ this.properties = checkNotNull(properties);
+ this.deliveryGuarantee = checkNotNull(deliveryGuarantee);
+ this.topicRouter = topicRouter;
+ this.topicRoutingMode = checkNotNull(topicRoutingMode);
+ this.messageDelayMillis = messageDelayMillis;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return this.changelogMode;
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+
+ final PulsarSerializationSchema<RowData> pulsarSerializationSchema =
+ serializationSchemaFactory.createPulsarSerializationSchema(context);
+
+ final PulsarSinkBuilder<RowData> pulsarSinkBuilder =
+ PulsarSink.builder()
+ .setSerializationSchema(pulsarSerializationSchema)
+ .setProperties(properties)
+ .setDeliveryGuarantee(deliveryGuarantee)
+ .setTopics(topics)
+ .setTopicRoutingMode(topicRoutingMode)
+ .delaySendingMessage(new FixedMessageDelayer<>(messageDelayMillis));
Review Comment:
Updated.
--
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-connector-pulsar] tisonkun commented on pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1691719534
@leonardBang All comments addressed. Also pushed a commit for including upsert mode code. Please give another look.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304186976
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
Review Comment:
We can file a JIRA ticket to support "topic-pattern" in the SQL connector. It's already supported by the DataStream connector.
--
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-connector-pulsar] tisonkun commented on pull request #56: [FLINK-XXXXX] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1653237834
> outside of the ASF
@MartijnVisser those patched are licensed with ALv2 also, and with the copyright of StreamNative Inc. where we agree to contribute it to the upstream.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304385180
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableValidationUtils.java:
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.RowKind;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.shade.com.google.common.collect.Sets;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FIELDS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_CUSTOM_TOPIC_ROUTER;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_TOPIC_ROUTING_MODE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AFTER_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.TOPICS;
+import static org.apache.pulsar.common.naming.TopicName.isValid;
+
+/** Util class for source and sink validation rules. */
+public class PulsarTableValidationUtils {
+
+ private PulsarTableValidationUtils() {}
+
+ public static void validatePrimaryKeyConstraints(
+ ObjectIdentifier tableName,
+ int[] primaryKeyIndexes,
+ FactoryUtil.TableFactoryHelper helper) {
+ final DecodingFormat<DeserializationSchema<RowData>> format =
+ getValueDecodingFormat(helper);
+ if (primaryKeyIndexes.length > 0
+ && format.getChangelogMode().containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ String.format(
+ "The Pulsar table '%s' with '%s' format doesn't support defining PRIMARY KEY constraint"
+ + " on the table, because it can't guarantee the semantic of primary key.",
+ tableName.asSummaryString(), format));
+ }
+ }
+
+ public static void validateTableSourceOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateStartCursorConfigs(tableOptions);
+ validateStopCursorConfigs(tableOptions);
+ validateSubscriptionTypeConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ }
+
+ public static void validateTableSinkOptions(ReadableConfig tableOptions) {
+ validateTopicsConfigs(tableOptions);
+ validateKeyFormatConfigs(tableOptions);
+ validateSinkRoutingConfigs(tableOptions);
+ }
+
+ protected static void validateTopicsConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.get(TOPICS).isEmpty()) {
+ throw new ValidationException("The topics list should not be empty.");
+ }
+
+ for (String topic : tableOptions.get(TOPICS)) {
+ if (!isValid(topic)) {
+ throw new ValidationException(
+ String.format("The topics name %s is not a valid topic name.", topic));
+ }
+ }
+ }
+
+ protected static void validateStartCursorConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SOURCE_START_FROM_MESSAGE_ID).isPresent()
+ && tableOptions.getOptional(SOURCE_START_FROM_PUBLISH_TIME).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SOURCE_START_FROM_MESSAGE_ID, SOURCE_START_FROM_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateStopCursorConfigs(ReadableConfig tableOptions) {
+ Set<ConfigOption<?>> conflictConfigOptions =
+ Sets.newHashSet(
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME);
+
+ long configsNums =
+ conflictConfigOptions.stream()
+ .map(tableOptions::getOptional)
+ .filter(Optional::isPresent)
+ .count();
+
+ if (configsNums > 1) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s, %s and %s can be specified. Detected more than 1 of them",
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME));
+ }
+ }
+
+ protected static void validateSubscriptionTypeConfigs(ReadableConfig tableOptions) {
+ SubscriptionType subscriptionType = tableOptions.get(SOURCE_SUBSCRIPTION_TYPE);
+
+ if (subscriptionType == SubscriptionType.Failover) {
+ throw new ValidationException(
+ String.format(
+ "%s SubscriptionType is not supported. ", SubscriptionType.Failover));
+ }
+ }
+
+ protected static void validateKeyFormatConfigs(ReadableConfig tableOptions) {
+ final Optional<String> optionalKeyFormat = tableOptions.getOptional(KEY_FORMAT);
+ final Optional<List<String>> optionalKeyFields = tableOptions.getOptional(KEY_FIELDS);
+ if (!optionalKeyFormat.isPresent() && optionalKeyFields.isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "The option '%s' can only be declared if a key format is defined using '%s'.",
+ KEY_FIELDS.key(), KEY_FORMAT.key()));
+ } else if (optionalKeyFormat.isPresent()
+ && (!optionalKeyFields.isPresent() || optionalKeyFields.get().size() == 0)) {
+ throw new ValidationException(
+ String.format(
+ "A key format '%s' requires the declaration of one or more of key fields using '%s'.",
+ KEY_FORMAT.key(), KEY_FIELDS.key()));
+ }
+ }
+
+ protected static void validateSinkRoutingConfigs(ReadableConfig tableOptions) {
+ if (tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).orElse(TopicRoutingMode.ROUND_ROBIN)
+ == TopicRoutingMode.CUSTOM) {
+ throw new ValidationException(
+ String.format(
+ "Only %s and %s can be used. For %s, please use sink.custom-topic-router for"
+ + "custom topic router and do not set this config.",
+ TopicRoutingMode.ROUND_ROBIN,
+ TopicRoutingMode.MESSAGE_KEY_HASH,
+ TopicRoutingMode.CUSTOM));
+ }
+ if (tableOptions.getOptional(SINK_CUSTOM_TOPIC_ROUTER).isPresent()
+ && tableOptions.getOptional(SINK_TOPIC_ROUTING_MODE).isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "Only one of %s and %s can be specified. Detected both of them",
+ SINK_CUSTOM_TOPIC_ROUTER, SINK_TOPIC_ROUTING_MODE));
+ }
+ }
+
+ protected static void validateUpsertModeKeyConstraints(
+ ReadableConfig tableOptions, int[] primaryKeyIndexes) {
+ if (!tableOptions.getOptional(KEY_FIELDS).isPresent()) {
+ throw new ValidationException(
+ "Upsert mode requires key.fields set to the primary key fields, should be set");
+ }
+
+ if (tableOptions.getOptional(KEY_FIELDS).get().size() == 0
+ || primaryKeyIndexes.length == 0) {
+ throw new ValidationException(
+ "'upsert-pulsar' require to define a PRIMARY KEY constraint. "
Review Comment:
Implemented.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304385684
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
Review Comment:
Update docs.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304176465
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
Review Comment:
I don't understand your suggestion. `Earliest` and `lastest` can be regarded as specific positions that the Pulsar broker can understand. They are barely positions that don't need a separate config option.
--
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-connector-pulsar] tisonkun commented on pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1655131338
Support keybytes for now to minimize code change. Following work -
1. Add SQL client E2E test
2. Add upsert table factory
3. Revisit key serialization support. I know that users can have biz data in keys, but I don't know if it's common. @leonardBang @PatrickRen does Kafka supports mapping message key to Flink SQL columns?
--
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-connector-pulsar] MartijnVisser commented on pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1653739866
> Because we need to take copyright considerations, licenses etc into account, before that code can be merged into an ASF project.
@tisonkun and I talked about this offline and checked it, there's no problem to accept these contributions
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304161739
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableFactory.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.pulsar.common.config.PulsarOptions;
+import org.apache.flink.connector.pulsar.sink.PulsarSinkOptions;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.table.sink.PulsarTableSerializationSchemaFactory;
+import org.apache.flink.connector.pulsar.table.sink.PulsarTableSink;
+import org.apache.flink.connector.pulsar.table.source.PulsarTableDeserializationSchemaFactory;
+import org.apache.flink.connector.pulsar.table.source.PulsarTableSource;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.createKeyFormatProjection;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.createValueFormatProjection;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getKeyDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getKeyEncodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getMessageDelayMillis;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getPulsarProperties;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getStartCursor;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getStopCursor;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getSubscriptionType;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getTopicListFromOptions;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getTopicRouter;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getTopicRoutingMode;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueDecodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptionUtils.getValueEncodingFormat;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.ADMIN_URL;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.EXPLICIT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FIELDS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.KEY_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SERVICE_URL;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_CUSTOM_TOPIC_ROUTER;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_MESSAGE_DELAY_INTERVAL;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SINK_TOPIC_ROUTING_MODE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_START_FROM_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AFTER_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_MESSAGE_ID;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_STOP_AT_PUBLISH_TIME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.SOURCE_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.TOPICS;
+import static org.apache.flink.connector.pulsar.table.PulsarTableOptions.VALUE_FORMAT;
+import static org.apache.flink.connector.pulsar.table.PulsarTableValidationUtils.validatePrimaryKeyConstraints;
+import static org.apache.flink.connector.pulsar.table.PulsarTableValidationUtils.validateTableSinkOptions;
+import static org.apache.flink.connector.pulsar.table.PulsarTableValidationUtils.validateTableSourceOptions;
+import static org.apache.flink.table.factories.FactoryUtil.SINK_PARALLELISM;
+import static org.apache.pulsar.shade.org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+
+/**
+ * Factory for creating {@link DynamicTableSource} and {@link DynamicTableSink}.
+ *
+ * <p>The main role of this class is to retrieve config options and validate options from config and
+ * the table schema. It also sets default values if a config option is not present.
+ */
+public class PulsarTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ public static final String IDENTIFIER = "pulsar";
+
+ public static final String DEFAULT_SUBSCRIPTION_NAME_PREFIX = "flink-sql-connector-pulsar-";
+
+ public static final boolean UPSERT_DISABLED = false;
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ // Format options should be retrieved before validation.
+ final DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat =
+ getKeyDecodingFormat(helper);
+ final DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat =
+ getValueDecodingFormat(helper);
+ ReadableConfig tableOptions = helper.getOptions();
+
+ // Validate configs are not conflict; each options is consumed; no unwanted configs
+ // PulsarOptions, PulsarSourceOptions and PulsarSinkOptions is not part of the validation.
+ helper.validateExcept(
+ PulsarOptions.CLIENT_CONFIG_PREFIX,
+ PulsarOptions.ADMIN_CONFIG_PREFIX,
+ PulsarSourceOptions.SOURCE_CONFIG_PREFIX,
+ PulsarSourceOptions.CONSUMER_CONFIG_PREFIX,
+ PulsarSinkOptions.PRODUCER_CONFIG_PREFIX,
+ PulsarSinkOptions.SINK_CONFIG_PREFIX);
+
+ validatePrimaryKeyConstraints(
+ context.getObjectIdentifier(), context.getPrimaryKeyIndexes(), helper);
+
+ validateTableSourceOptions(tableOptions);
+
+ // Retrieve configs
+ final List<String> topics = getTopicListFromOptions(tableOptions);
+ final StartCursor startCursor = getStartCursor(tableOptions);
+ final StopCursor stopCursor = getStopCursor(tableOptions);
+ final SubscriptionType subscriptionType = getSubscriptionType(tableOptions);
+
+ // Forward source configs
+ final Properties properties = getPulsarProperties(tableOptions);
+ properties.setProperty(PULSAR_ADMIN_URL.key(), tableOptions.get(ADMIN_URL));
+ properties.setProperty(PULSAR_SERVICE_URL.key(), tableOptions.get(SERVICE_URL));
+ // Set random subscriptionName if not provided
+ properties.setProperty(
+ PULSAR_SUBSCRIPTION_NAME.key(),
+ tableOptions
+ .getOptional(SOURCE_SUBSCRIPTION_NAME)
+ .orElse(DEFAULT_SUBSCRIPTION_NAME_PREFIX + randomAlphabetic(5)));
+ // Retrieve physical fields (not including computed or metadata fields),
+ // and projections and create a schema factory based on such information.
+ final DataType physicalDataType = context.getPhysicalRowDataType();
+
+ final int[] valueProjection = createValueFormatProjection(tableOptions, physicalDataType);
+ final int[] keyProjection = createKeyFormatProjection(tableOptions, physicalDataType);
+
+ final PulsarTableDeserializationSchemaFactory deserializationSchemaFactory =
+ new PulsarTableDeserializationSchemaFactory(
+ physicalDataType,
+ keyDecodingFormat,
+ keyProjection,
+ valueDecodingFormat,
+ valueProjection,
+ UPSERT_DISABLED);
+
+ // Set default values for configuration not exposed to user.
+ final DecodingFormat<DeserializationSchema<RowData>> decodingFormatForMetadataPushdown =
+ valueDecodingFormat;
+ final ChangelogMode changelogMode = decodingFormatForMetadataPushdown.getChangelogMode();
+
+ return new PulsarTableSource(
+ deserializationSchemaFactory,
+ decodingFormatForMetadataPushdown,
+ changelogMode,
+ topics,
+ properties,
+ startCursor,
+ stopCursor,
+ subscriptionType);
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ // Format options should be retrieved before validation.
+ final EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat =
+ getKeyEncodingFormat(helper);
+ final EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat =
+ getValueEncodingFormat(helper);
+ ReadableConfig tableOptions = helper.getOptions();
+
+ // Validate configs are not conflict; each options is consumed; no unwanted configs
+ // PulsarOptions, PulsarSourceOptions and PulsarSinkOptions is not part of the validation.
+ helper.validateExcept(
+ PulsarOptions.CLIENT_CONFIG_PREFIX,
+ PulsarOptions.ADMIN_CONFIG_PREFIX,
+ PulsarSourceOptions.SOURCE_CONFIG_PREFIX,
+ PulsarSourceOptions.CONSUMER_CONFIG_PREFIX,
+ PulsarSinkOptions.PRODUCER_CONFIG_PREFIX,
+ PulsarSinkOptions.SINK_CONFIG_PREFIX);
+
+ validatePrimaryKeyConstraints(
+ context.getObjectIdentifier(), context.getPrimaryKeyIndexes(), helper);
+
+ validateTableSinkOptions(tableOptions);
+
+ // Retrieve configs
+ final TopicRouter<RowData> topicRouter =
+ getTopicRouter(tableOptions, context.getClassLoader());
+ final TopicRoutingMode topicRoutingMode = getTopicRoutingMode(tableOptions);
+ final long messageDelayMillis = getMessageDelayMillis(tableOptions);
+
+ final List<String> topics = getTopicListFromOptions(tableOptions);
+
+ // Forward sink configs
+ final Properties properties = getPulsarProperties(tableOptions);
+ properties.setProperty(PULSAR_ADMIN_URL.key(), tableOptions.get(ADMIN_URL));
+ properties.setProperty(PULSAR_SERVICE_URL.key(), tableOptions.get(SERVICE_URL));
+
+ // Retrieve physical DataType (not including computed or metadata fields)
+ final DataType physicalDataType = context.getPhysicalRowDataType();
+ final int[] keyProjection = createKeyFormatProjection(tableOptions, physicalDataType);
+ final int[] valueProjection = createValueFormatProjection(tableOptions, physicalDataType);
+
+ final PulsarTableSerializationSchemaFactory serializationSchemaFactory =
+ new PulsarTableSerializationSchemaFactory(
+ physicalDataType,
+ keyEncodingFormat,
+ keyProjection,
+ valueEncodingFormat,
+ valueProjection,
+ UPSERT_DISABLED);
+
+ // Set default values for configuration not exposed to user.
+ final DeliveryGuarantee deliveryGuarantee = DeliveryGuarantee.AT_LEAST_ONCE;
+ final ChangelogMode changelogMode = valueEncodingFormat.getChangelogMode();
+
+ return new PulsarTableSink(
+ serializationSchemaFactory,
+ changelogMode,
+ topics,
+ properties,
+ deliveryGuarantee,
+ topicRouter,
+ topicRoutingMode,
+ messageDelayMillis);
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Stream.of(TOPICS, ADMIN_URL, SERVICE_URL).collect(Collectors.toSet());
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Stream.of(
+ FactoryUtil.FORMAT,
+ VALUE_FORMAT,
+ SOURCE_SUBSCRIPTION_NAME,
+ SOURCE_SUBSCRIPTION_TYPE,
+ SOURCE_START_FROM_MESSAGE_ID,
+ SOURCE_START_FROM_PUBLISH_TIME,
+ SOURCE_STOP_AT_MESSAGE_ID,
+ SOURCE_STOP_AFTER_MESSAGE_ID,
+ SOURCE_STOP_AT_PUBLISH_TIME,
+ SINK_CUSTOM_TOPIC_ROUTER,
+ SINK_TOPIC_ROUTING_MODE,
+ SINK_MESSAGE_DELAY_INTERVAL,
+ SINK_PARALLELISM,
+ KEY_FORMAT,
+ KEY_FIELDS,
+ EXPLICIT)
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * Format and Delivery guarantee related options are not forward options.
+ *
+ * @return
Review Comment:
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-connector-pulsar] syhily commented on pull request #56: [FLINK-XXXXX] Basic table factory for Pulsar connector
Posted by "syhily (via GitHub)" <gi...@apache.org>.
syhily commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1649158056
This is a huge PR. I may have time this weekend to glance at 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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304162911
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarRowDataConverter.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.DeserializationException;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.Collector;
+
+import org.apache.pulsar.client.api.Message;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Contains the projection information needed to map a Pulsar message to proper key fields, value
+ * fields and metadata fields.
+ */
+public class PulsarRowDataConverter implements Serializable {
+ private static final long serialVersionUID = -3399264407634977459L;
Review Comment:
Fixed.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchemaFactory.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.source;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.table.connector.Projection;
+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.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Contains key, value projection and format information, and use such information to create a
+ * {@link PulsarTableDeserializationSchema} instance used by runtime {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSource} instance.
+ *
+ * <p>A Flink row fields has a strict order: Physical Fields (Key + value) + Format Metadata Fields
+ * Connector Metadata Fields. Physical Fields are fields come directly from Pulsar message body;
+ * Format Metadata Fields are from the extra information from the decoding format. Connector
+ * metadata fields are the ones most Pulsar messages have, such as publish time, message size and
+ * producer name.
+ *
+ * <p>In general, Physical fields + Format Metadata fields are contained in the RowData decoded
+ * using valueDecodingFormat. Only Connector Metadata fields needs to be appended to the decoded
+ * RowData. The tricky part is to put format metadata and connector metadata in the right location.
+ * This requires an explicit adjustment process.
+ *
+ * <p>For example, suppose Physical Fields (Key + value) + Format Metadata Fields + Connector
+ * Metadata Fields. has arity of 11, key projection is [0, 6], and physical value projection is [1,
+ * 2, 3, 4, 5], Then after the adjustment, key projection should be [0, 6], physical value
+ * projection should be [1, 2, 3, 4, 5] and format metadata projection should be [7], connector
+ * metadata projection should be [8, 9, 10].
+ */
+public class PulsarTableDeserializationSchemaFactory implements Serializable {
+
+ private static final long serialVersionUID = 6091562041940740434L;
Review Comment:
Fixed.
--
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-connector-pulsar] nlu90 commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "nlu90 (via GitHub)" <gi...@apache.org>.
nlu90 commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1301892005
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
+ + "source. Use \"never\", \"latest\" or pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\"");
+
+ public static final ConfigOption<String> SOURCE_STOP_AFTER_MESSAGE_ID =
+ ConfigOptions.key("source.stop.after-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop position but include the "
+ + "given message in the consuming result for the unbounded sql "
+ + "source. Pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\". ");
+
+ public static final ConfigOption<Long> SOURCE_STOP_AT_PUBLISH_TIME =
+ ConfigOptions.key("source.stop.at-publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional publish timestamp used to specify a stop cursor"
+ + " for the unbounded sql source.");
+
+ // --------------------------------------------------------------------------------------------
+ // Table Sink Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> SINK_CUSTOM_TOPIC_ROUTER =
+ ConfigOptions.key("sink.custom-topic-router")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the custom topic router class URL that is used in the [Pulsar DataStream sink connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-sink). If this option is provided, the %s option will be ignored.",
+ code("sink.topic-routing-mode"))
+ .build());
+
+ public static final ConfigOption<TopicRoutingMode> SINK_TOPIC_ROUTING_MODE =
+ ConfigOptions.key("sink.topic-routing-mode")
+ .enumType(TopicRoutingMode.class)
+ .defaultValue(TopicRoutingMode.ROUND_ROBIN)
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the topic routing mode. Available options are %s and %s. By default, it is set to %s. If you want to use a custom topic router, use the %s option to determine the partition for a particular message.",
+ code("round-robin"),
+ code("message-key-hash"),
+ code("round-robin"),
+ code("sink.custom-topic-router"))
+ .build());
+
+ public static final ConfigOption<Duration> SINK_MESSAGE_DELAY_INTERVAL =
+ ConfigOptions.key("sink.message-delay-interval")
+ .durationType()
+ .defaultValue(Duration.ZERO)
+ .withDescription(
+ "(Optional) the message delay delivery interval that is used in the [Pulsar DataStream sink connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-sink).");
+
+ // --------------------------------------------------------------------------------------------
+ // Format Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> KEY_FORMAT =
+ ConfigOptions.key("key" + FORMAT_SUFFIX)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the format identifier for decoding/encoding key bytes in "
+ + "Pulsar message. The identifier is used to discover a suitable format factory.");
+
+ public static final ConfigOption<List<String>> KEY_FIELDS =
+ ConfigOptions.key("key.fields")
+ .stringType()
+ .asList()
+ .defaultValues()
+ .withDescription(
+ "An explicit list of physical columns from the table schema that are decoded/encoded from the key bytes of a Pulsar message. By default, this list is empty and thus a key is undefined.");
+
+ public static final ConfigOption<String> VALUE_FORMAT =
+ ConfigOptions.key("value" + FORMAT_SUFFIX)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the format identifier for decoding/encoding value data. "
+ + "The identifier is used to discover a suitable format factory.");
+
+ // --------------------------------------------------------------------------------------------
+ // Pulsar Options
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions#PULSAR_ADMIN_URL}. Copied here
+ * because it is a required config option and should not be included in the {@link
+ * org.apache.flink.table.factories.FactoryUtil.FactoryHelper#validateExcept(String...)} method.
+ *
+ * <p>By default all {@link org.apache.flink.connector.pulsar.common.config.PulsarOptions} are
+ * included in the validateExcept() method./p>
+ */
+ public static final ConfigOption<String> ADMIN_URL =
Review Comment:
we should look into removing the admin url
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304172096
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
+ + "source. Use \"never\", \"latest\" or pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\"");
+
+ public static final ConfigOption<String> SOURCE_STOP_AFTER_MESSAGE_ID =
+ ConfigOptions.key("source.stop.after-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop position but include the "
+ + "given message in the consuming result for the unbounded sql "
+ + "source. Pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\". ");
+
+ public static final ConfigOption<Long> SOURCE_STOP_AT_PUBLISH_TIME =
+ ConfigOptions.key("source.stop.at-publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional publish timestamp used to specify a stop cursor"
+ + " for the unbounded sql source.");
Review Comment:
No. They are called at `setUnboundedStopCursor` which contains `this.boundedness = Boundedness.CONTINUOUS_UNBOUNDED;`
cc @syhily may provide more context why we support UNBOUNDED with stop cursor in the underneath connector. Since a "stop" semantic is to have bounded process.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
Review Comment:
Ditto https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304172096
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1304164358
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSerializationSchema.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * Licensed 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.
+ */
+
+package org.apache.flink.connector.pulsar.table.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
+import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
+import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessage;
+import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessageBuilder;
+import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
+
+import org.apache.pulsar.client.api.Schema;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@link PulsarSerializationSchema} implementation for Pulsar SQL sink connector. It is
+ * responsible for retrieving fields from Flink row and serialize into Pulsar message key or body,
+ * and set necessary metadata fields as required.
+ */
+public class PulsarTableSerializationSchema implements PulsarSerializationSchema<RowData> {
+
+ private static final long serialVersionUID = 7314442107082067836L;
Review Comment:
Fixed.
--
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-connector-pulsar] tisonkun commented on a diff in pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#discussion_r1302557629
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.pulsar.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT_SUFFIX;
+
+/**
+ * Config options that is used to configure a Pulsar SQL Connector. These config options are
+ * specific to SQL Connectors only. Other runtime configurations can be found in {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions}, {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions}, and {@link
+ * org.apache.flink.connector.pulsar.sink.PulsarSinkOptions}.
+ */
+@PublicEvolving
+public final class PulsarTableOptions {
+
+ private PulsarTableOptions() {}
+
+ public static final ConfigOption<List<String>> TOPICS =
+ ConfigOptions.key("topics")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Topic name(s) the table reads data from. It can be a single topic name or a list of topic names separated by a semicolon symbol (%s) like %s.",
+ code(";"), code("topic-1;topic-2"))
+ .build());
+
+ // --------------------------------------------------------------------------------------------
+ // Table Source Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<SubscriptionType> SOURCE_SUBSCRIPTION_TYPE =
+ ConfigOptions.key("source.subscription-type")
+ .enumType(SubscriptionType.class)
+ .defaultValue(SubscriptionType.Exclusive)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+ code("Exclusive"), code("Shared"))
+ .build());
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+ * Copied because we want to have a default value for it.
+ */
+ public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+ ConfigOptions.key("source.subscription-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The subscription name of the consumer that is used by the runtime [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). This argument is required for constructing the consumer.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_START_FROM_MESSAGE_ID =
+ ConfigOptions.key("source.start.message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Optional message id used to specify a consuming starting point for "
+ + "source. Use %s, %s or pass in a message id "
+ + "representation in %s, "
+ + "such as %s",
+ code("earliest"),
+ code("latest"),
+ code("ledgerId:entryId:partitionId"),
+ code("12:2:-1"))
+ .build());
+
+ public static final ConfigOption<Long> SOURCE_START_FROM_PUBLISH_TIME =
+ ConfigOptions.key("source.start.publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the publish timestamp that is used to specify a starting point for the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source) to consume data.")
+ .build());
+
+ public static final ConfigOption<String> SOURCE_STOP_AT_MESSAGE_ID =
+ ConfigOptions.key("source.stop.at-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop cursor for the unbounded sql "
+ + "source. Use \"never\", \"latest\" or pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\"");
+
+ public static final ConfigOption<String> SOURCE_STOP_AFTER_MESSAGE_ID =
+ ConfigOptions.key("source.stop.after-message-id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional message id used to specify a stop position but include the "
+ + "given message in the consuming result for the unbounded sql "
+ + "source. Pass in a message id "
+ + "representation in \"ledgerId:entryId:partitionId\", "
+ + "such as \"12:2:-1\". ");
+
+ public static final ConfigOption<Long> SOURCE_STOP_AT_PUBLISH_TIME =
+ ConfigOptions.key("source.stop.at-publish-time")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional publish timestamp used to specify a stop cursor"
+ + " for the unbounded sql source.");
+
+ // --------------------------------------------------------------------------------------------
+ // Table Sink Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> SINK_CUSTOM_TOPIC_ROUTER =
+ ConfigOptions.key("sink.custom-topic-router")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the custom topic router class URL that is used in the [Pulsar DataStream sink connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-sink). If this option is provided, the %s option will be ignored.",
+ code("sink.topic-routing-mode"))
+ .build());
+
+ public static final ConfigOption<TopicRoutingMode> SINK_TOPIC_ROUTING_MODE =
+ ConfigOptions.key("sink.topic-routing-mode")
+ .enumType(TopicRoutingMode.class)
+ .defaultValue(TopicRoutingMode.ROUND_ROBIN)
+ .withDescription(
+ Description.builder()
+ .text(
+ "(Optional) the topic routing mode. Available options are %s and %s. By default, it is set to %s. If you want to use a custom topic router, use the %s option to determine the partition for a particular message.",
+ code("round-robin"),
+ code("message-key-hash"),
+ code("round-robin"),
+ code("sink.custom-topic-router"))
+ .build());
+
+ public static final ConfigOption<Duration> SINK_MESSAGE_DELAY_INTERVAL =
+ ConfigOptions.key("sink.message-delay-interval")
+ .durationType()
+ .defaultValue(Duration.ZERO)
+ .withDescription(
+ "(Optional) the message delay delivery interval that is used in the [Pulsar DataStream sink connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-sink).");
+
+ // --------------------------------------------------------------------------------------------
+ // Format Options
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> KEY_FORMAT =
+ ConfigOptions.key("key" + FORMAT_SUFFIX)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the format identifier for decoding/encoding key bytes in "
+ + "Pulsar message. The identifier is used to discover a suitable format factory.");
+
+ public static final ConfigOption<List<String>> KEY_FIELDS =
+ ConfigOptions.key("key.fields")
+ .stringType()
+ .asList()
+ .defaultValues()
+ .withDescription(
+ "An explicit list of physical columns from the table schema that are decoded/encoded from the key bytes of a Pulsar message. By default, this list is empty and thus a key is undefined.");
+
+ public static final ConfigOption<String> VALUE_FORMAT =
+ ConfigOptions.key("value" + FORMAT_SUFFIX)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the format identifier for decoding/encoding value data. "
+ + "The identifier is used to discover a suitable format factory.");
+
+ // --------------------------------------------------------------------------------------------
+ // Pulsar Options
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Exactly same as {@link
+ * org.apache.flink.connector.pulsar.common.config.PulsarOptions#PULSAR_ADMIN_URL}. Copied here
+ * because it is a required config option and should not be included in the {@link
+ * org.apache.flink.table.factories.FactoryUtil.FactoryHelper#validateExcept(String...)} method.
+ *
+ * <p>By default all {@link org.apache.flink.connector.pulsar.common.config.PulsarOptions} are
+ * included in the validateExcept() method./p>
+ */
+ public static final ConfigOption<String> ADMIN_URL =
Review Comment:
Yeah. I saw your issue here - https://issues.apache.org/jira/browse/FLINK-32938.
This is an inherited issue that we should handle it separately.
--
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-connector-pulsar] MartijnVisser commented on pull request #56: [FLINK-XXXXX] Basic table factory for Pulsar connector
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1653220914
> I'll respect all their credits before merging.
@tisonkun This seems to indicate that this contains code that's been contributed by people outside of the ASF, is that true? Because we need to take copyright considerations, licenses etc into account, before that code can be merged into an ASF project.
--
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-connector-pulsar] tisonkun commented on pull request #56: [FLINK-26203] Basic table factory for Pulsar connector
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #56:
URL: https://github.com/apache/flink-connector-pulsar/pull/56#issuecomment-1653298964
Update JIRA number, this should cover several subtasks under FLINK-26203.
--
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