@Test public void recover() throws IOException { put(); localFileStore.forceFlush(); ConcurrentMap<ConsumePartition, Position> recover = localFileStore.recover(); Position positionVal = recover.get(consumePartition); Assert.assertEquals(position.toString(), positionVal.toString()); }
public ConcurrentMap<ConsumePartition, Position> recover() throws IOException { ConcurrentMap<ConsumePartition, Position> consumePositionCache = new ConcurrentHashMap<>(); Map<Joint, List<ConsumeBill>> consumeBills; try { consumeBills = loadFromFile(indexFile, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } catch (Exception e) { consumeBills = loadFromFile(indexFileBack, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } if (consumeBills != null) { consumeBills.entrySet().stream().forEach(entry -> { Joint key = entry.getKey(); entry.getValue().stream().forEach(val -> { ConsumePartition consumePartition = new ConsumePartition(key.getTopic(), key.getApp(), val.getPartition()); consumePartition.setPartitionGroup(val.getPartitionGroup()); consumePositionCache.putIfAbsent(consumePartition, new Position(val.getAckStartIndex(), val.getAckCurIndex(), val.getPullStartIndex(), val.getPullCurIndex())); } ); }); } return consumePositionCache; }
LocalFileStore implements PositionStore<ConsumePartition, Position> { public ConcurrentMap<ConsumePartition, Position> recover() throws IOException { ConcurrentMap<ConsumePartition, Position> consumePositionCache = new ConcurrentHashMap<>(); Map<Joint, List<ConsumeBill>> consumeBills; try { consumeBills = loadFromFile(indexFile, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } catch (Exception e) { consumeBills = loadFromFile(indexFileBack, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } if (consumeBills != null) { consumeBills.entrySet().stream().forEach(entry -> { Joint key = entry.getKey(); entry.getValue().stream().forEach(val -> { ConsumePartition consumePartition = new ConsumePartition(key.getTopic(), key.getApp(), val.getPartition()); consumePartition.setPartitionGroup(val.getPartitionGroup()); consumePositionCache.putIfAbsent(consumePartition, new Position(val.getAckStartIndex(), val.getAckCurIndex(), val.getPullStartIndex(), val.getPullCurIndex())); } ); }); } return consumePositionCache; } }
LocalFileStore implements PositionStore<ConsumePartition, Position> { public ConcurrentMap<ConsumePartition, Position> recover() throws IOException { ConcurrentMap<ConsumePartition, Position> consumePositionCache = new ConcurrentHashMap<>(); Map<Joint, List<ConsumeBill>> consumeBills; try { consumeBills = loadFromFile(indexFile, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } catch (Exception e) { consumeBills = loadFromFile(indexFileBack, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } if (consumeBills != null) { consumeBills.entrySet().stream().forEach(entry -> { Joint key = entry.getKey(); entry.getValue().stream().forEach(val -> { ConsumePartition consumePartition = new ConsumePartition(key.getTopic(), key.getApp(), val.getPartition()); consumePartition.setPartitionGroup(val.getPartitionGroup()); consumePositionCache.putIfAbsent(consumePartition, new Position(val.getAckStartIndex(), val.getAckCurIndex(), val.getPullStartIndex(), val.getPullCurIndex())); } ); }); } return consumePositionCache; }  }
LocalFileStore implements PositionStore<ConsumePartition, Position> { public ConcurrentMap<ConsumePartition, Position> recover() throws IOException { ConcurrentMap<ConsumePartition, Position> consumePositionCache = new ConcurrentHashMap<>(); Map<Joint, List<ConsumeBill>> consumeBills; try { consumeBills = loadFromFile(indexFile, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } catch (Exception e) { consumeBills = loadFromFile(indexFileBack, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } if (consumeBills != null) { consumeBills.entrySet().stream().forEach(entry -> { Joint key = entry.getKey(); entry.getValue().stream().forEach(val -> { ConsumePartition consumePartition = new ConsumePartition(key.getTopic(), key.getApp(), val.getPartition()); consumePartition.setPartitionGroup(val.getPartitionGroup()); consumePositionCache.putIfAbsent(consumePartition, new Position(val.getAckStartIndex(), val.getAckCurIndex(), val.getPullStartIndex(), val.getPullCurIndex())); } ); }); } return consumePositionCache; }  void setBasePath(String basePath); @Override void start(); @Override void stop(); @Override boolean isStarted(); @Override Position get(ConsumePartition key); @Override void put(ConsumePartition key, Position value); @Override Position remove(ConsumePartition key); @Override Position putIfAbsent(ConsumePartition key, Position value); @Override void forceFlush(); @Override Iterator<ConsumePartition> iterator(); ConcurrentMap<ConsumePartition, Position> recover(); }
LocalFileStore implements PositionStore<ConsumePartition, Position> { public ConcurrentMap<ConsumePartition, Position> recover() throws IOException { ConcurrentMap<ConsumePartition, Position> consumePositionCache = new ConcurrentHashMap<>(); Map<Joint, List<ConsumeBill>> consumeBills; try { consumeBills = loadFromFile(indexFile, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } catch (Exception e) { consumeBills = loadFromFile(indexFileBack, new TypeReference<Map<Joint, List<ConsumeBill>>>() { }); } if (consumeBills != null) { consumeBills.entrySet().stream().forEach(entry -> { Joint key = entry.getKey(); entry.getValue().stream().forEach(val -> { ConsumePartition consumePartition = new ConsumePartition(key.getTopic(), key.getApp(), val.getPartition()); consumePartition.setPartitionGroup(val.getPartitionGroup()); consumePositionCache.putIfAbsent(consumePartition, new Position(val.getAckStartIndex(), val.getAckCurIndex(), val.getPullStartIndex(), val.getPullCurIndex())); } ); }); } return consumePositionCache; }  void setBasePath(String basePath); @Override void start(); @Override void stop(); @Override boolean isStarted(); @Override Position get(ConsumePartition key); @Override void put(ConsumePartition key, Position value); @Override Position remove(ConsumePartition key); @Override Position putIfAbsent(ConsumePartition key, Position value); @Override void forceFlush(); @Override Iterator<ConsumePartition> iterator(); ConcurrentMap<ConsumePartition, Position> recover();  }