### Question:
FermaFlowPathRepository extends FermaGenericRepository<FlowPath, FlowPathData, FlowPathFrame> implements FlowPathRepository { @Override public Collection<FlowPath> findActualByFlowIds(Set<String> flowIds) { Set<String> pathIds = new HashSet<>(); framedGraph().traverse(g -> g.V() .hasLabel(FlowFrame.FRAME_LABEL) .has(FlowFrame.FLOW_ID_PROPERTY, P.within(flowIds)) .values(FlowFrame.FORWARD_PATH_ID_PROPERTY, FlowFrame.REVERSE_PATH_ID_PROPERTY, FlowFrame.PROTECTED_FORWARD_PATH_ID_PROPERTY, FlowFrame.PROTECTED_REVERSE_PATH_ID_PROPERTY)) .getRawTraversal() .forEachRemaining(pathId -> pathIds.add((String) pathId)); return framedGraph().traverse(g -> g.V() .hasLabel(FlowPathFrame.FRAME_LABEL) .has(FlowPathFrame.PATH_ID_PROPERTY, P.within(pathIds))) .toListExplicit(FlowPathFrame.class).stream() .map(FlowPath::new) .collect(Collectors.toList()); } FermaFlowPathRepository(FramedGraphFactory<?> graphFactory, TransactionManager transactionManager); @Override Collection<FlowPath> findAll(); @Override Optional<FlowPath> findById(PathId pathId); @Override Optional<FlowPath> findByFlowIdAndCookie(String flowId, FlowSegmentCookie cookie); @Override Collection<FlowPath> findByFlowId(String flowId); @Override Collection<FlowPath> findByFlowGroupId(String flowGroupId); @Override Collection<PathId> findPathIdsByFlowGroupId(String flowGroupId); @Override Collection<FlowPath> findActualByFlowIds(Set<String> flowIds); @Override Collection<FlowPath> findBySrcSwitch(SwitchId switchId, boolean includeProtected); @Override Collection<FlowPath> findByEndpointSwitch(SwitchId switchId, boolean includeProtected); @Override Collection<FlowPath> findBySegmentSwitch(SwitchId switchId); @Override Collection<FlowPath> findInactiveBySegmentSwitch(SwitchId switchId); @Override Collection<FlowPath> findBySegmentSwitchWithMultiTable(SwitchId switchId, boolean multiTable); @Override Collection<FlowPath> findWithPathSegment(SwitchId srcSwitchId, int srcPort,
                                                    SwitchId dstSwitchId, int dstPort); @Override Collection<FlowPath> findBySegmentDestSwitch(SwitchId switchId); @Override Collection<FlowPath> findBySegmentEndpoint(SwitchId switchId, int port); @Override void updateStatus(PathId pathId, FlowPathStatus pathStatus); @Override long getUsedBandwidthBetweenEndpoints(SwitchId srcSwitchId, int srcPort, SwitchId dstSwitchId, int dstPort); @Override Optional<FlowPath> remove(PathId pathId);  }### Answer:
@Test public void shouldFindFlowPathIdsByFlowIds() { Flow flowA = buildTestProtectedFlow(TEST_FLOW_ID_1, switchA, PORT_1, VLAN_1, switchB, PORT_2, VLAN_2); flowRepository.add(flowA); Flow flowB = buildTestFlow(TEST_FLOW_ID_2, switchA, PORT_1, VLAN_2, switchB, PORT_2, 0); flowRepository.add(flowB); Flow flowC = buildTestProtectedFlow(TEST_FLOW_ID_3, switchB, PORT_1, VLAN_1, switchB, PORT_3, VLAN_1); flowRepository.add(flowC); Collection<FlowPath> flowPaths = flowPathRepository.findActualByFlowIds(Sets.newHashSet(TEST_FLOW_ID_1, TEST_FLOW_ID_2)); Collection<PathId> pathIds = flowPaths.stream().map(FlowPath::getPathId).collect(Collectors.toList()); assertEquals(6, pathIds.size()); assertTrue(pathIds.contains(flowA.getForwardPathId())); assertTrue(pathIds.contains(flowA.getReversePathId())); assertTrue(pathIds.contains(flowA.getProtectedForwardPathId())); assertTrue(pathIds.contains(flowA.getProtectedReversePathId())); assertTrue(pathIds.contains(flowB.getForwardPathId())); assertTrue(pathIds.contains(flowB.getReversePathId())); }