@Test @SneakyThrows public void shouldSelectOnlyFeatureSetsThatJobManagerSubscribedTo() { Store store = DataGenerator.getDefaultStore(); Source source = DataGenerator.getDefaultSource(); FeatureSet featureSet1 = DataGenerator.createFeatureSet(source, "default", "fs1"); FeatureSet featureSet2 = DataGenerator.createFeatureSet(source, "project", "fs3"); FeatureSet featureSet3 = DataGenerator.createFeatureSet(source, "default", "not-fs"); when(specService.listFeatureSets( CoreServiceProto.ListFeatureSetsRequest.newBuilder() .setFilter(Filter.newBuilder().setFeatureSetName("*").setProject("*").build()) .build())) .thenReturn( ListFeatureSetsResponse.newBuilder() .addAllFeatureSets(Lists.newArrayList(featureSet1, featureSet2, featureSet3)) .build()); List<FeatureSet> featureSetsForStore = controllerWithConsolidation.getFeatureSetsForStore(store); assertThat(featureSetsForStore, containsInAnyOrder(featureSet1, featureSet2)); }
List<FeatureSet> getFeatureSetsForStore(Store store) { return store.getSubscriptionsList().stream() .flatMap( subscription -> { return specService .listFeatureSets( CoreServiceProto.ListFeatureSetsRequest.newBuilder() .setFilter( CoreServiceProto.ListFeatureSetsRequest.Filter.newBuilder() .setProject(subscription.getProject()) .setFeatureSetName(subscription.getName()) .build()) .build()) .getFeatureSetsList().stream() .filter( f -> this.featureSetSubscriptions.isEmpty() || isSubscribedToFeatureSet( this.featureSetSubscriptions, f.getSpec().getProject(), f.getSpec().getName())); }) .distinct() .collect(Collectors.toList()); }
JobControllerService { List<FeatureSet> getFeatureSetsForStore(Store store) { return store.getSubscriptionsList().stream() .flatMap( subscription -> { return specService .listFeatureSets( CoreServiceProto.ListFeatureSetsRequest.newBuilder() .setFilter( CoreServiceProto.ListFeatureSetsRequest.Filter.newBuilder() .setProject(subscription.getProject()) .setFeatureSetName(subscription.getName()) .build()) .build()) .getFeatureSetsList().stream() .filter( f -> this.featureSetSubscriptions.isEmpty() || isSubscribedToFeatureSet( this.featureSetSubscriptions, f.getSpec().getProject(), f.getSpec().getName())); }) .distinct() .collect(Collectors.toList()); } }
JobControllerService { List<FeatureSet> getFeatureSetsForStore(Store store) { return store.getSubscriptionsList().stream() .flatMap( subscription -> { return specService .listFeatureSets( CoreServiceProto.ListFeatureSetsRequest.newBuilder() .setFilter( CoreServiceProto.ListFeatureSetsRequest.Filter.newBuilder() .setProject(subscription.getProject()) .setFeatureSetName(subscription.getName()) .build()) .build()) .getFeatureSetsList().stream() .filter( f -> this.featureSetSubscriptions.isEmpty() || isSubscribedToFeatureSet( this.featureSetSubscriptions, f.getSpec().getProject(), f.getSpec().getName())); }) .distinct() .collect(Collectors.toList()); } @Autowired  JobControllerService(
      JobRepository jobRepository,
      CoreServiceGrpc.CoreServiceBlockingStub specService,
      JobManager jobManager,
      FeastProperties feastProperties,
      JobGroupingStrategy groupingStrategy,
      KafkaTemplate<String, FeatureSetSpec> specPublisher); }
JobControllerService { List<FeatureSet> getFeatureSetsForStore(Store store) { return store.getSubscriptionsList().stream() .flatMap( subscription -> { return specService .listFeatureSets( CoreServiceProto.ListFeatureSetsRequest.newBuilder() .setFilter( CoreServiceProto.ListFeatureSetsRequest.Filter.newBuilder() .setProject(subscription.getProject()) .setFeatureSetName(subscription.getName()) .build()) .build()) .getFeatureSetsList().stream() .filter( f -> this.featureSetSubscriptions.isEmpty() || isSubscribedToFeatureSet( this.featureSetSubscriptions, f.getSpec().getProject(), f.getSpec().getName())); }) .distinct() .collect(Collectors.toList()); } @Autowired  JobControllerService(
      JobRepository jobRepository,
      CoreServiceGrpc.CoreServiceBlockingStub specService,
      JobManager jobManager,
      FeastProperties feastProperties,
      JobGroupingStrategy groupingStrategy,
      KafkaTemplate<String, FeatureSetSpec> specPublisher); @Scheduled(fixedDelayString = "${feast.jobs.polling_interval_milliseconds}") void Poll(); @Scheduled(fixedDelayString = "${feast.stream.specsOptions.notifyIntervalMilliseconds}") void notifyJobsWhenFeatureSetUpdated(); @KafkaListener( topics = {"${feast.stream.specsOptions.specsAckTopic}"}, containerFactory = "kafkaAckListenerContainerFactory") void listenAckFromJobs(
      ConsumerRecord<String, IngestionJobProto.FeatureSetSpecAck> record); }
JobControllerService { List<FeatureSet> getFeatureSetsForStore(Store store) { return store.getSubscriptionsList().stream() .flatMap( subscription -> { return specService .listFeatureSets( CoreServiceProto.ListFeatureSetsRequest.newBuilder() .setFilter( CoreServiceProto.ListFeatureSetsRequest.Filter.newBuilder() .setProject(subscription.getProject()) .setFeatureSetName(subscription.getName()) .build()) .build()) .getFeatureSetsList().stream() .filter( f -> this.featureSetSubscriptions.isEmpty() || isSubscribedToFeatureSet( this.featureSetSubscriptions, f.getSpec().getProject(), f.getSpec().getName())); }) .distinct() .collect(Collectors.toList()); } @Autowired  JobControllerService(
      JobRepository jobRepository,
      CoreServiceGrpc.CoreServiceBlockingStub specService,
      JobManager jobManager,
      FeastProperties feastProperties,
      JobGroupingStrategy groupingStrategy,
      KafkaTemplate<String, FeatureSetSpec> specPublisher); @Scheduled(fixedDelayString = "${feast.jobs.polling_interval_milliseconds}") void Poll(); @Scheduled(fixedDelayString = "${feast.stream.specsOptions.notifyIntervalMilliseconds}") void notifyJobsWhenFeatureSetUpdated(); @KafkaListener( topics = {"${feast.stream.specsOptions.specsAckTopic}"}, containerFactory = "kafkaAckListenerContainerFactory") void listenAckFromJobs(
      ConsumerRecord<String, IngestionJobProto.FeatureSetSpecAck> record); static final String VERSION_LABEL; }