@Test public void testNullTimezone() { TimeBasedPartitioner<String> partitioner = new TimeBasedPartitioner<>(); Map<String, Object> config = createConfig(null); final String configKey = PartitionerConfig.TIMEZONE_CONFIG; final String msg = "Timezone cannot be empty."; String timeZoneString = null; config.put(configKey, timeZoneString); Exception e = assertThrows(ConfigException.class, () -> { partitioner.configure(config); }); assertEquals( String.format("Invalid value %s for configuration %s: " + msg, timeZoneString, configKey), e.getMessage()); }
@Override public void configure(Map<String, Object> config) { super.configure(config); long partitionDurationMsProp = (long) config.get(PartitionerConfig.PARTITION_DURATION_MS_CONFIG); if (partitionDurationMsProp < 0) { throw new ConfigException( PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMsProp, "Partition duration needs to be a positive." ); } String pathFormat = (String) config.get(PartitionerConfig.PATH_FORMAT_CONFIG); if (StringUtils.isBlank(pathFormat) || pathFormat.equals(delim)) { throw new ConfigException( PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat, "Path format cannot be empty." ); } else if (!StringUtils.isBlank(delim) && pathFormat.endsWith(delim)) { pathFormat = pathFormat.substring(0, pathFormat.length() - delim.length()); } String localeString = (String) config.get(PartitionerConfig.LOCALE_CONFIG); if (StringUtils.isBlank(localeString)) { throw new ConfigException( PartitionerConfig.LOCALE_CONFIG, localeString, "Locale cannot be empty." ); } String timeZoneString = (String) config.get(PartitionerConfig.TIMEZONE_CONFIG); if (StringUtils.isBlank(timeZoneString)) { throw new ConfigException( PartitionerConfig.TIMEZONE_CONFIG, timeZoneString, "Timezone cannot be empty." ); } Locale locale = new Locale(localeString); DateTimeZone timeZone = DateTimeZone.forID(timeZoneString); init(partitionDurationMsProp, pathFormat, locale, timeZone, config); }
TimeBasedPartitioner extends DefaultPartitioner<T> { @Override public void configure(Map<String, Object> config) { super.configure(config); long partitionDurationMsProp = (long) config.get(PartitionerConfig.PARTITION_DURATION_MS_CONFIG); if (partitionDurationMsProp < 0) { throw new ConfigException( PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMsProp, "Partition duration needs to be a positive." ); } String pathFormat = (String) config.get(PartitionerConfig.PATH_FORMAT_CONFIG); if (StringUtils.isBlank(pathFormat) || pathFormat.equals(delim)) { throw new ConfigException( PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat, "Path format cannot be empty." ); } else if (!StringUtils.isBlank(delim) && pathFormat.endsWith(delim)) { pathFormat = pathFormat.substring(0, pathFormat.length() - delim.length()); } String localeString = (String) config.get(PartitionerConfig.LOCALE_CONFIG); if (StringUtils.isBlank(localeString)) { throw new ConfigException( PartitionerConfig.LOCALE_CONFIG, localeString, "Locale cannot be empty." ); } String timeZoneString = (String) config.get(PartitionerConfig.TIMEZONE_CONFIG); if (StringUtils.isBlank(timeZoneString)) { throw new ConfigException( PartitionerConfig.TIMEZONE_CONFIG, timeZoneString, "Timezone cannot be empty." ); } Locale locale = new Locale(localeString); DateTimeZone timeZone = DateTimeZone.forID(timeZoneString); init(partitionDurationMsProp, pathFormat, locale, timeZone, config); } }
TimeBasedPartitioner extends DefaultPartitioner<T> { @Override public void configure(Map<String, Object> config) { super.configure(config); long partitionDurationMsProp = (long) config.get(PartitionerConfig.PARTITION_DURATION_MS_CONFIG); if (partitionDurationMsProp < 0) { throw new ConfigException( PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMsProp, "Partition duration needs to be a positive." ); } String pathFormat = (String) config.get(PartitionerConfig.PATH_FORMAT_CONFIG); if (StringUtils.isBlank(pathFormat) || pathFormat.equals(delim)) { throw new ConfigException( PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat, "Path format cannot be empty." ); } else if (!StringUtils.isBlank(delim) && pathFormat.endsWith(delim)) { pathFormat = pathFormat.substring(0, pathFormat.length() - delim.length()); } String localeString = (String) config.get(PartitionerConfig.LOCALE_CONFIG); if (StringUtils.isBlank(localeString)) { throw new ConfigException( PartitionerConfig.LOCALE_CONFIG, localeString, "Locale cannot be empty." ); } String timeZoneString = (String) config.get(PartitionerConfig.TIMEZONE_CONFIG); if (StringUtils.isBlank(timeZoneString)) { throw new ConfigException( PartitionerConfig.TIMEZONE_CONFIG, timeZoneString, "Timezone cannot be empty." ); } Locale locale = new Locale(localeString); DateTimeZone timeZone = DateTimeZone.forID(timeZoneString); init(partitionDurationMsProp, pathFormat, locale, timeZone, config); }  }
TimeBasedPartitioner extends DefaultPartitioner<T> { @Override public void configure(Map<String, Object> config) { super.configure(config); long partitionDurationMsProp = (long) config.get(PartitionerConfig.PARTITION_DURATION_MS_CONFIG); if (partitionDurationMsProp < 0) { throw new ConfigException( PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMsProp, "Partition duration needs to be a positive." ); } String pathFormat = (String) config.get(PartitionerConfig.PATH_FORMAT_CONFIG); if (StringUtils.isBlank(pathFormat) || pathFormat.equals(delim)) { throw new ConfigException( PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat, "Path format cannot be empty." ); } else if (!StringUtils.isBlank(delim) && pathFormat.endsWith(delim)) { pathFormat = pathFormat.substring(0, pathFormat.length() - delim.length()); } String localeString = (String) config.get(PartitionerConfig.LOCALE_CONFIG); if (StringUtils.isBlank(localeString)) { throw new ConfigException( PartitionerConfig.LOCALE_CONFIG, localeString, "Locale cannot be empty." ); } String timeZoneString = (String) config.get(PartitionerConfig.TIMEZONE_CONFIG); if (StringUtils.isBlank(timeZoneString)) { throw new ConfigException( PartitionerConfig.TIMEZONE_CONFIG, timeZoneString, "Timezone cannot be empty." ); } Locale locale = new Locale(localeString); DateTimeZone timeZone = DateTimeZone.forID(timeZoneString); init(partitionDurationMsProp, pathFormat, locale, timeZone, config); }  static long getPartition(long timeGranularityMs, long timestamp, DateTimeZone timeZone); long getPartitionDurationMs(); String getPathFormat(); TimestampExtractor getTimestampExtractor(); @Override void configure(Map<String, Object> config); @Override String encodePartition(SinkRecord sinkRecord, long nowInMillis); @Override String encodePartition(SinkRecord sinkRecord); @Override List<T> partitionFields(); TimestampExtractor newTimestampExtractor(String extractorClassName); }
TimeBasedPartitioner extends DefaultPartitioner<T> { @Override public void configure(Map<String, Object> config) { super.configure(config); long partitionDurationMsProp = (long) config.get(PartitionerConfig.PARTITION_DURATION_MS_CONFIG); if (partitionDurationMsProp < 0) { throw new ConfigException( PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMsProp, "Partition duration needs to be a positive." ); } String pathFormat = (String) config.get(PartitionerConfig.PATH_FORMAT_CONFIG); if (StringUtils.isBlank(pathFormat) || pathFormat.equals(delim)) { throw new ConfigException( PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat, "Path format cannot be empty." ); } else if (!StringUtils.isBlank(delim) && pathFormat.endsWith(delim)) { pathFormat = pathFormat.substring(0, pathFormat.length() - delim.length()); } String localeString = (String) config.get(PartitionerConfig.LOCALE_CONFIG); if (StringUtils.isBlank(localeString)) { throw new ConfigException( PartitionerConfig.LOCALE_CONFIG, localeString, "Locale cannot be empty." ); } String timeZoneString = (String) config.get(PartitionerConfig.TIMEZONE_CONFIG); if (StringUtils.isBlank(timeZoneString)) { throw new ConfigException( PartitionerConfig.TIMEZONE_CONFIG, timeZoneString, "Timezone cannot be empty." ); } Locale locale = new Locale(localeString); DateTimeZone timeZone = DateTimeZone.forID(timeZoneString); init(partitionDurationMsProp, pathFormat, locale, timeZone, config); }  static long getPartition(long timeGranularityMs, long timestamp, DateTimeZone timeZone); long getPartitionDurationMs(); String getPathFormat(); TimestampExtractor getTimestampExtractor(); @Override void configure(Map<String, Object> config); @Override String encodePartition(SinkRecord sinkRecord, long nowInMillis); @Override String encodePartition(SinkRecord sinkRecord); @Override List<T> partitionFields(); TimestampExtractor newTimestampExtractor(String extractorClassName);  }