code: @Override
  	public Map<String, String> toProperties() {
  		Map<String, String> properties = new HashMap<>();
  		properties.put(Rowtime.ROWTIME_WATERMARKS_TYPE, Rowtime.ROWTIME_WATERMARKS_TYPE_VALUE_CUSTOM);
  		properties.put(Rowtime.ROWTIME_WATERMARKS_CLASS, this.getClass().getName());
  		properties.put(Rowtime.ROWTIME_WATERMARKS_SERIALIZED, EncodingUtils.encodeObjectToString(this));
  		return properties;
  	}
docs: This method is a default implementation that uses java serialization and it is discouraged.
  All implementation should provide a more specific set of properties.