@Test public void noMatchForPartition() throws IOException { Assert.assertEquals(OffsetComputer.NO_OFFSET, offsetComputer.computeOffsets(Collections.singleton(123)).get(123).longValue()); }
@Override public Map<Integer, Long> computeOffsets(Collection<Integer> partitionIds) throws IOException { Set<Integer> partitionIdsSet = new HashSet<>(partitionIds); final Map<Integer, Long> result = partitionIds.stream().collect(Collectors.toMap(Function.identity(), i -> NO_OFFSET)); final Map<Integer, Map<String, FinalEventPartitionFile>> resultFile = new HashMap<>(); LocalDateTime today = LocalDateTime.now(); for (int i = 0; i < backlogDays; ++i) { LocalDateTime day = today.minusDays(i); String listedDay = day.format(DateTimeFormatter.ISO_DATE); String dirsPattern = String.format("%s/*", computeDirName(day)); FileStatus[] fileStatuses = fs.globStatus(new Path(basePath, dirsPattern)); for (FileStatus status : fileStatuses) { String fileName = status.getPath().getName(); Matcher matcher = offsetFilePatternGenerator.matcher(fileName); if (matcher.matches()) { try { int partitionId = Integer.parseInt(matcher.group("partitionId")); long index = Long.parseLong(matcher.group("index")); if (isPartitionComputedByThisReader(partitionId, partitionIdsSet) && checkCurrentFileIndexBiggerTheOneInHashPartitionDay(partitionId, listedDay, index, resultFile)) { HashMap<String, FinalEventPartitionFile> dateFinalEventPartitionFile = (HashMap<String, FinalEventPartitionFile>) resultFile.computeIfAbsent(partitionId, HashMap::new); dateFinalEventPartitionFile.put(listedDay, new FinalEventPartitionFile(index, status.getPath())); resultFile.put(partitionId, dateFinalEventPartitionFile); } } catch (NumberFormatException e) { LOGGER.info("Couldn't deviate a valid offset from '{}'", fileName); } } } } for (Map.Entry<Integer, Map<String, FinalEventPartitionFile>> partitionIdDateFile : resultFile.entrySet()) { result.put(partitionIdDateFile.getKey(), getMaxOffset(partitionIdDateFile.getValue())); } return result; }
HdfsOffsetComputer implements OffsetComputer { @Override public Map<Integer, Long> computeOffsets(Collection<Integer> partitionIds) throws IOException { Set<Integer> partitionIdsSet = new HashSet<>(partitionIds); final Map<Integer, Long> result = partitionIds.stream().collect(Collectors.toMap(Function.identity(), i -> NO_OFFSET)); final Map<Integer, Map<String, FinalEventPartitionFile>> resultFile = new HashMap<>(); LocalDateTime today = LocalDateTime.now(); for (int i = 0; i < backlogDays; ++i) { LocalDateTime day = today.minusDays(i); String listedDay = day.format(DateTimeFormatter.ISO_DATE); String dirsPattern = String.format("%s/*", computeDirName(day)); FileStatus[] fileStatuses = fs.globStatus(new Path(basePath, dirsPattern)); for (FileStatus status : fileStatuses) { String fileName = status.getPath().getName(); Matcher matcher = offsetFilePatternGenerator.matcher(fileName); if (matcher.matches()) { try { int partitionId = Integer.parseInt(matcher.group("partitionId")); long index = Long.parseLong(matcher.group("index")); if (isPartitionComputedByThisReader(partitionId, partitionIdsSet) && checkCurrentFileIndexBiggerTheOneInHashPartitionDay(partitionId, listedDay, index, resultFile)) { HashMap<String, FinalEventPartitionFile> dateFinalEventPartitionFile = (HashMap<String, FinalEventPartitionFile>) resultFile.computeIfAbsent(partitionId, HashMap::new); dateFinalEventPartitionFile.put(listedDay, new FinalEventPartitionFile(index, status.getPath())); resultFile.put(partitionId, dateFinalEventPartitionFile); } } catch (NumberFormatException e) { LOGGER.info("Couldn't deviate a valid offset from '{}'", fileName); } } } } for (Map.Entry<Integer, Map<String, FinalEventPartitionFile>> partitionIdDateFile : resultFile.entrySet()) { result.put(partitionIdDateFile.getKey(), getMaxOffset(partitionIdDateFile.getValue())); } return result; } }
HdfsOffsetComputer implements OffsetComputer { @Override public Map<Integer, Long> computeOffsets(Collection<Integer> partitionIds) throws IOException { Set<Integer> partitionIdsSet = new HashSet<>(partitionIds); final Map<Integer, Long> result = partitionIds.stream().collect(Collectors.toMap(Function.identity(), i -> NO_OFFSET)); final Map<Integer, Map<String, FinalEventPartitionFile>> resultFile = new HashMap<>(); LocalDateTime today = LocalDateTime.now(); for (int i = 0; i < backlogDays; ++i) { LocalDateTime day = today.minusDays(i); String listedDay = day.format(DateTimeFormatter.ISO_DATE); String dirsPattern = String.format("%s/*", computeDirName(day)); FileStatus[] fileStatuses = fs.globStatus(new Path(basePath, dirsPattern)); for (FileStatus status : fileStatuses) { String fileName = status.getPath().getName(); Matcher matcher = offsetFilePatternGenerator.matcher(fileName); if (matcher.matches()) { try { int partitionId = Integer.parseInt(matcher.group("partitionId")); long index = Long.parseLong(matcher.group("index")); if (isPartitionComputedByThisReader(partitionId, partitionIdsSet) && checkCurrentFileIndexBiggerTheOneInHashPartitionDay(partitionId, listedDay, index, resultFile)) { HashMap<String, FinalEventPartitionFile> dateFinalEventPartitionFile = (HashMap<String, FinalEventPartitionFile>) resultFile.computeIfAbsent(partitionId, HashMap::new); dateFinalEventPartitionFile.put(listedDay, new FinalEventPartitionFile(index, status.getPath())); resultFile.put(partitionId, dateFinalEventPartitionFile); } } catch (NumberFormatException e) { LOGGER.info("Couldn't deviate a valid offset from '{}'", fileName); } } } } for (Map.Entry<Integer, Map<String, FinalEventPartitionFile>> partitionIdDateFile : resultFile.entrySet()) { result.put(partitionIdDateFile.getKey(), getMaxOffset(partitionIdDateFile.getValue())); } return result; } HdfsOffsetComputer(FileSystem fs, Path basePath, int backlogDays);  HdfsOffsetComputer(FileSystem fs, Path basePath, @Nullable String kafkaCluster, int backlogDays); }
HdfsOffsetComputer implements OffsetComputer { @Override public Map<Integer, Long> computeOffsets(Collection<Integer> partitionIds) throws IOException { Set<Integer> partitionIdsSet = new HashSet<>(partitionIds); final Map<Integer, Long> result = partitionIds.stream().collect(Collectors.toMap(Function.identity(), i -> NO_OFFSET)); final Map<Integer, Map<String, FinalEventPartitionFile>> resultFile = new HashMap<>(); LocalDateTime today = LocalDateTime.now(); for (int i = 0; i < backlogDays; ++i) { LocalDateTime day = today.minusDays(i); String listedDay = day.format(DateTimeFormatter.ISO_DATE); String dirsPattern = String.format("%s/*", computeDirName(day)); FileStatus[] fileStatuses = fs.globStatus(new Path(basePath, dirsPattern)); for (FileStatus status : fileStatuses) { String fileName = status.getPath().getName(); Matcher matcher = offsetFilePatternGenerator.matcher(fileName); if (matcher.matches()) { try { int partitionId = Integer.parseInt(matcher.group("partitionId")); long index = Long.parseLong(matcher.group("index")); if (isPartitionComputedByThisReader(partitionId, partitionIdsSet) && checkCurrentFileIndexBiggerTheOneInHashPartitionDay(partitionId, listedDay, index, resultFile)) { HashMap<String, FinalEventPartitionFile> dateFinalEventPartitionFile = (HashMap<String, FinalEventPartitionFile>) resultFile.computeIfAbsent(partitionId, HashMap::new); dateFinalEventPartitionFile.put(listedDay, new FinalEventPartitionFile(index, status.getPath())); resultFile.put(partitionId, dateFinalEventPartitionFile); } } catch (NumberFormatException e) { LOGGER.info("Couldn't deviate a valid offset from '{}'", fileName); } } } } for (Map.Entry<Integer, Map<String, FinalEventPartitionFile>> partitionIdDateFile : resultFile.entrySet()) { result.put(partitionIdDateFile.getKey(), getMaxOffset(partitionIdDateFile.getValue())); } return result; } HdfsOffsetComputer(FileSystem fs, Path basePath, int backlogDays);  HdfsOffsetComputer(FileSystem fs, Path basePath, @Nullable String kafkaCluster, int backlogDays); @Override Map<Integer, Long> computeOffsets(Collection<Integer> partitionIds); @Override String computeDirName(LocalDateTime time); @Override long getIndex(String fileName); @Override String computeTopicGlob(LocalDateTime time, int partition); @Override String computePath(LocalDateTime time, long index, int partition); }
HdfsOffsetComputer implements OffsetComputer { @Override public Map<Integer, Long> computeOffsets(Collection<Integer> partitionIds) throws IOException { Set<Integer> partitionIdsSet = new HashSet<>(partitionIds); final Map<Integer, Long> result = partitionIds.stream().collect(Collectors.toMap(Function.identity(), i -> NO_OFFSET)); final Map<Integer, Map<String, FinalEventPartitionFile>> resultFile = new HashMap<>(); LocalDateTime today = LocalDateTime.now(); for (int i = 0; i < backlogDays; ++i) { LocalDateTime day = today.minusDays(i); String listedDay = day.format(DateTimeFormatter.ISO_DATE); String dirsPattern = String.format("%s/*", computeDirName(day)); FileStatus[] fileStatuses = fs.globStatus(new Path(basePath, dirsPattern)); for (FileStatus status : fileStatuses) { String fileName = status.getPath().getName(); Matcher matcher = offsetFilePatternGenerator.matcher(fileName); if (matcher.matches()) { try { int partitionId = Integer.parseInt(matcher.group("partitionId")); long index = Long.parseLong(matcher.group("index")); if (isPartitionComputedByThisReader(partitionId, partitionIdsSet) && checkCurrentFileIndexBiggerTheOneInHashPartitionDay(partitionId, listedDay, index, resultFile)) { HashMap<String, FinalEventPartitionFile> dateFinalEventPartitionFile = (HashMap<String, FinalEventPartitionFile>) resultFile.computeIfAbsent(partitionId, HashMap::new); dateFinalEventPartitionFile.put(listedDay, new FinalEventPartitionFile(index, status.getPath())); resultFile.put(partitionId, dateFinalEventPartitionFile); } } catch (NumberFormatException e) { LOGGER.info("Couldn't deviate a valid offset from '{}'", fileName); } } } } for (Map.Entry<Integer, Map<String, FinalEventPartitionFile>> partitionIdDateFile : resultFile.entrySet()) { result.put(partitionIdDateFile.getKey(), getMaxOffset(partitionIdDateFile.getValue())); } return result; } HdfsOffsetComputer(FileSystem fs, Path basePath, int backlogDays);  HdfsOffsetComputer(FileSystem fs, Path basePath, @Nullable String kafkaCluster, int backlogDays); @Override Map<Integer, Long> computeOffsets(Collection<Integer> partitionIds); @Override String computeDirName(LocalDateTime time); @Override long getIndex(String fileName); @Override String computeTopicGlob(LocalDateTime time, int partition); @Override String computePath(LocalDateTime time, long index, int partition);  }