[ 
https://issues.apache.org/jira/browse/FLINK-8538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16374585#comment-16374585
 ] 

ASF GitHub Bot commented on FLINK-8538:
---------------------------------------

Github user fhueske commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5564#discussion_r170232160
  
    --- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/DescriptorProperties.scala
 ---
    @@ -246,13 +394,93 @@ class DescriptorProperties(normalizeKeys: Boolean = 
true) {
         Some(schemaBuilder.build())
       }
     
    +  /**
    +    * Returns a table schema under the given key if it exists.
    +    */
    +  def getOptionalTableSchema(key: String): Optional[TableSchema] = 
toJava(getTableSchema(key))
    +
    +  /**
    +    * Returns the type information under the given key if it exists.
    +    */
    +  def getType(key: String): Option[TypeInformation[_]] = {
    +    properties.get(key).map(TypeStringUtils.readTypeInfo)
    +  }
    +
    +  /**
    +    * Returns the type information under the given key if it exists.
    +    * This method is intended for Java code.
    +    */
    +  def getOptionalType(key: String): Optional[TypeInformation[_]] = {
    +    toJava(getType(key))
    +  }
    +
    +  /**
    +    * Returns a prefix subset of properties.
    +    */
    +  def getPrefix(prefixKey: String): Map[String, String] = {
    +    val prefix = prefixKey + '.'
    +    properties.filterKeys(_.startsWith(prefix)).toSeq.map{ case (k, v) =>
    +      k.substring(prefix.length) -> v // remove prefix
    +    }.toMap
    +  }
    +
    +  /**
    +    * Returns a prefix subset of properties.
    +    * This method is intended for Java code.
    +    */
    +  def getPrefixMap(prefixKey: String): JMap[String, String] = 
getPrefix(prefixKey).asJava
    --- End diff --
    
    I find the different names for methods that do the same confusing. I'd just 
remove the Scala methods.


> Add a Kafka table source factory with JSON format support
> ---------------------------------------------------------
>
>                 Key: FLINK-8538
>                 URL: https://issues.apache.org/jira/browse/FLINK-8538
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Table API & SQL
>            Reporter: Timo Walther
>            Assignee: Xingcan Cui
>            Priority: Major
>
> Similar to CSVTableSourceFactory a Kafka table source factory for JSON should 
> be added. This issue includes improving the existing JSON descriptor with 
> validation that can be used for other connectors as well. It is up for 
> discussion if we want to split the KafkaJsonTableSource into connector and 
> format such that we can reuse the format for other table sources as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Reply via email to