fix pulsar admin auth for PulsarProtobufNativeFormatFactory (#391)

(cherry picked from commit d682122)

pulsar-flink-connector/src/main/java/org/apache/flink/formats/protobufnative/PulsarProtobufNativeFormatFactory.java

@@ -19,6 +19,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.streaming.connectors.pulsar.internal.PulsarClientUtils;
+import org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions;
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.format.DecodingFormat;
@@ -41,12 +42,10 @@
 
 import java.util.Collections;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.ADMIN_URL;
-import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.PROPERTIES;
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.TOPIC;
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.TOPIC_PATTERN;
 
@@ -65,12 +64,7 @@ public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(Dynam
 
         String topic = extractTopicName(context);
         String adminUrl = tableConf.get(ADMIN_URL);
-        Optional<Map<String, String>> stringMap = tableConf.getOptional(PROPERTIES);
-        Properties properties = stringMap.map((map) -> {
-            final Properties prop = new Properties();
-            prop.putAll(map);
-            return prop;
-        }).orElse(new Properties());
+        Properties properties = PulsarTableOptions.getPulsarProperties(context.getCatalogTable().getOptions());
 
         SerializableSupplier<Descriptors.Descriptor> loadDescriptor = () -> {
             SchemaInfo schemaInfo = null;