@Test public void testGeneratePartitionedPath() throws Exception { BiHourlyPartitioner partitioner = (BiHourlyPartitioner) configurePartitioner( new BiHourlyPartitioner(), null, null); SinkRecord sinkRecord = getSinkRecord(); String encodedPartition = partitioner.encodePartition(sinkRecord); final String topic = "topic"; String path = partitioner.generatePartitionedPath(topic, encodedPartition); assertEquals(topic+"/year=2015/month=4/day=2/hour=0/", path); }
@Override public String encodePartition(SinkRecord sinkRecord, long nowInMillis) { Long timestamp = timestampExtractor.extract(sinkRecord, nowInMillis); return encodedPartitionForTimestamp(sinkRecord, timestamp); }
TimeBasedPartitioner extends DefaultPartitioner<T> { @Override public String encodePartition(SinkRecord sinkRecord, long nowInMillis) { Long timestamp = timestampExtractor.extract(sinkRecord, nowInMillis); return encodedPartitionForTimestamp(sinkRecord, timestamp); } }
TimeBasedPartitioner extends DefaultPartitioner<T> { @Override public String encodePartition(SinkRecord sinkRecord, long nowInMillis) { Long timestamp = timestampExtractor.extract(sinkRecord, nowInMillis); return encodedPartitionForTimestamp(sinkRecord, timestamp); }  }
TimeBasedPartitioner extends DefaultPartitioner<T> { @Override public String encodePartition(SinkRecord sinkRecord, long nowInMillis) { Long timestamp = timestampExtractor.extract(sinkRecord, nowInMillis); return encodedPartitionForTimestamp(sinkRecord, timestamp); }  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 String encodePartition(SinkRecord sinkRecord, long nowInMillis) { Long timestamp = timestampExtractor.extract(sinkRecord, nowInMillis); return encodedPartitionForTimestamp(sinkRecord, timestamp); }  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);  }