flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
/**
		 * Configures the Kafka timestamp to be a rowtime attribute.
		 *
		 * <p>Note: Kafka supports message timestamps only since version 0.10.</p>
		 *
		 * @param rowtimeAttribute The name of the rowtime attribute in the table schema.
		 * @param watermarkStrategy The {@link WatermarkStrategy} to generate watermarks for the rowtime attribute.
		 * @return The builder.
		 */
Configures the Kafka timestamp to be a rowtime attribute.
Note: Kafka supports message timestamps only since version 0.10.

@param rowtimeAttribute The name of the rowtime attribute in the table schema.
@param watermarkStrategy The  WatermarkStrategy to generate watermarks for the rowtime attribute.
@return The builder.
Configures the Kafka timestamp to be a rowtime attribute . Note : Kafka supports message timestamps only since version 0 . 10 . @param rowtimeAttribute The name of the rowtime attribute in the table schema . @param watermarkStrategy The WatermarkStrategy to generate watermarks for the rowtime attribute . @return The builder .
public B withKafkaTimestampAsRowtimeAttribute(
				String rowtimeAttribute,
				WatermarkStrategy watermarkStrategy) {
			Preconditions.checkNotNull(rowtimeAttribute, "Rowtime attribute must not be null.");
			Preconditions.checkArgument(!rowtimeAttribute.isEmpty(), "Rowtime attribute must not be empty.");
			Preconditions.checkNotNull(watermarkStrategy, "Watermark assigner must not be null.");
			Preconditions.checkArgument(this.rowtimeAttributeDescriptor == null,
				"Currently, only one rowtime attribute is supported.");
			Preconditions.checkArgument(supportsKafkaTimestamps(), "Kafka timestamps are only supported since Kafka 0.10.");
			this.rowtimeAttributeDescriptor = new RowtimeAttributeDescriptor(
				rowtimeAttribute,
				new StreamRecordTimestamp(),
				watermarkStrategy);
			return builder();
		}
public B withKafkaTimestampAsRowtimeAttribute ( String rowtimeAttribute , WatermarkStrategy watermarkStrategy ) { Preconditions . checkNotNull ( rowtimeAttribute , "Rowtime attribute must not be null." ) ; Preconditions . checkArgument ( ! rowtimeAttribute . isEmpty ( ) , "Rowtime attribute must not be empty." ) ; Preconditions . checkNotNull ( watermarkStrategy , "Watermark assigner must not be null." ) ; Preconditions . checkArgument ( this . rowtimeAttributeDescriptor == null , "Currently, only one rowtime attribute is supported." ) ; Preconditions . checkArgument ( supportsKafkaTimestamps ( ) , "Kafka timestamps are only supported since Kafka 0.10." ) ; this . rowtimeAttributeDescriptor = new RowtimeAttributeDescriptor ( rowtimeAttribute , new StreamRecordTimestamp ( ) , watermarkStrategy ) ; return builder ( ) ; }
Configures the Kafka timestamp to be a rowtime attribute.
Configures the Kafka timestamp to be a rowtime attribute .