RedisBackedJobService implements JobService { @Override public Optional<Job> get(String id) { Job job = null; try { String json = new String(syncCommand.get(id.getBytes())); if (json.isEmpty()) { return Optional.empty(); } Builder builder = Job.newBuilder(); JsonFormat.parser().merge(json, builder); job = builder.build(); } catch (Exception e) { log.error(String.format("Failed to parse JSON for Feast job: %s", e.getMessage())); } return Optional.ofNullable(job); } RedisBackedJobService(FeastProperties.JobStoreProperties jobStoreProperties);  RedisBackedJobService(StatefulRedisConnection<byte[], byte[]> connection); @Override Optional<Job> get(String id); @Override void upsert(Job job);  }
@Test public void shouldRecoverIfRedisConnectionIsLost() { RedisClient client = RedisClient.create(RedisURI.create("localhost", REDIS_PORT)); RedisBackedJobService jobService = new RedisBackedJobService(client.connect(new ByteArrayCodec())); jobService.get("does not exist"); redis.stop(); try { jobService.get("does not exist"); } catch (Exception e) { } redis.start(); jobService.get("does not exist"); client.shutdown(); }