flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
/**
		 * Configures a field of the table to be a rowtime attribute.
		 * The configured field must be present in the tabel schema and of type {@link Types#SQL_TIMESTAMP()}.
		 *
		 * @param rowtimeAttribute The name of the rowtime attribute in the table schema.
		 * @param timestampExtractor The {@link TimestampExtractor} to extract the rowtime attribute from the physical type.
		 * @param watermarkStrategy The {@link WatermarkStrategy} to generate watermarks for the rowtime attribute.
		 * @return The builder.
		 */
Configures a field of the table to be a rowtime attribute.
The configured field must be present in the tabel schema and of type  Types#SQL_TIMESTAMP().

@param rowtimeAttribute The name of the rowtime attribute in the table schema.
@param timestampExtractor The  TimestampExtractor to extract the rowtime attribute from the physical type.
@param watermarkStrategy The  WatermarkStrategy to generate watermarks for the rowtime attribute.
@return The builder.
Configures a field of the table to be a rowtime attribute . The configured field must be present in the tabel schema and of type Types#SQL_TIMESTAMP () . @param rowtimeAttribute The name of the rowtime attribute in the table schema . @param timestampExtractor The TimestampExtractor to extract the rowtime attribute from the physical type . @param watermarkStrategy The WatermarkStrategy to generate watermarks for the rowtime attribute . @return The builder .
public B withRowtimeAttribute(
				String rowtimeAttribute,
				TimestampExtractor timestampExtractor,
				WatermarkStrategy watermarkStrategy) {
			Preconditions.checkNotNull(rowtimeAttribute, "Rowtime attribute must not be null.");
			Preconditions.checkArgument(!rowtimeAttribute.isEmpty(), "Rowtime attribute must not be empty.");
			Preconditions.checkNotNull(timestampExtractor, "Timestamp extractor must not be null.");
			Preconditions.checkNotNull(watermarkStrategy, "Watermark assigner must not be null.");
			Preconditions.checkArgument(this.rowtimeAttributeDescriptor == null,
				"Currently, only one rowtime attribute is supported.");
			this.rowtimeAttributeDescriptor = new RowtimeAttributeDescriptor(
				rowtimeAttribute,
				timestampExtractor,
				watermarkStrategy);
			return builder();
		}
public B withRowtimeAttribute ( String rowtimeAttribute , TimestampExtractor timestampExtractor , WatermarkStrategy watermarkStrategy ) { Preconditions . checkNotNull ( rowtimeAttribute , "Rowtime attribute must not be null." ) ; Preconditions . checkArgument ( ! rowtimeAttribute . isEmpty ( ) , "Rowtime attribute must not be empty." ) ; Preconditions . checkNotNull ( timestampExtractor , "Timestamp extractor must not be null." ) ; Preconditions . checkNotNull ( watermarkStrategy , "Watermark assigner must not be null." ) ; Preconditions . checkArgument ( this . rowtimeAttributeDescriptor == null , "Currently, only one rowtime attribute is supported." ) ; this . rowtimeAttributeDescriptor = new RowtimeAttributeDescriptor ( rowtimeAttribute , timestampExtractor , watermarkStrategy ) ; return builder ( ) ; }
Configures a field of the table to be a rowtime attribute.
Configures a field of the table to be a rowtime attribute .