### Question:
SparkLivyRestClient implements SparkShellRestClient { @VisibleForTesting Statement pollStatement(JerseyRestClient jerseyClient, SparkShellProcess sparkShellProcess, Integer stmtId) { Validate.isInstanceOf(SparkLivyProcess.class, sparkShellProcess, "SparkLivyRestClient.pollStatement called on non Livy Process"); SparkLivyProcess sparkLivyProcess = (SparkLivyProcess) sparkShellProcess; return livyClient.pollStatement(jerseyClient, sparkLivyProcess, stmtId); }  @Override SparkJobResponse createJob(@Nonnull final SparkShellProcess process, @Nonnull final SparkJobRequest request); @Nonnull @Override Optional<Response> downloadQuery(@Nonnull SparkShellProcess process, @Nonnull String queryId, @Nonnull String saveId); @Nonnull @Override Optional<TransformResponse> getQueryResult(@Nonnull SparkShellProcess process, @Nonnull String id); @Nonnull @Override Optional<SparkJobResponse> getJobResult(@Nonnull final SparkShellProcess process, @Nonnull final String id); @Nonnull @Override Optional<SaveResponse> getQuerySave(@Nonnull SparkShellProcess process, @Nonnull String
        queryId, @Nonnull String saveId); @Nonnull @Override TransformResponse query(@Nonnull SparkShellProcess process, @Nonnull TransformRequest request); @Nonnull @Override SaveResponse saveQuery(@Nonnull SparkShellProcess process, @Nonnull String id, @Nonnull SaveRequest
        request); @Nonnull @Override Optional<Response> downloadTransform(@Nonnull SparkShellProcess process, @Nonnull String transformId, @Nonnull String saveId); @Nonnull @Override DataSources getDataSources(@Nonnull SparkShellProcess process); @Nonnull @Override Optional<TransformResponse> getTransformResult(@Nonnull SparkShellProcess process, @Nonnull String
        transformId); @Nonnull @Override Optional<SaveResponse> getTransformSave(@Nonnull SparkShellProcess process, @Nonnull String
        transformId, @Nonnull String saveId); @Nonnull @Override SaveResponse saveTransform(@Nonnull SparkShellProcess process, @Nonnull String
        transformId, @Nonnull SaveRequest request); @Nonnull @Override TransformResponse transform(@Nonnull SparkShellProcess process, @Nonnull TransformRequest request); @Nonnull TransformResponse kyloCatalogTransform(@Nonnull final SparkShellProcess process, @Nonnull final KyloCatalogReadRequest request); @Nonnull @Override ServerStatusResponse serverStatus(SparkShellProcess sparkShellProcess); @Resource
    public FileSystem fileSystem; }### Answer:
@Test @Ignore public void testGetLivySessions() throws JsonProcessingException, InterruptedException { System.setProperty("sun.security.krb5.debug", "true"); System.setProperty("sun.security.jgss.debug", "true"); SessionsGet sg = new SessionsGet.Builder().from(1).size(2).build(); String sgJson = new ObjectMapper().writeValueAsString(sg); logger.debug("{}", sgJson); assertThat(sgJson).isEqualToIgnoringCase("{\"from\":1,\"size\":2}"); logger.debug("kerberosSparkProperties={}", kerberosSparkProperties); SparkShellProcess sparkProcess = sparkLivyProcessManager.getProcessForUser("kylo"); JerseyRestClient client = sparkLivyProcessManager.getClient(sparkProcess); sparkLivyProcessManager.start("kylo"); Integer stmtId = 0; Statement statement = livyRestProvider.pollStatement(client, sparkProcess, stmtId); logger.debug("statement={}", statement); assertThat(statement.getState()).isEqualTo(StatementState.available); assertThat(statement.getOutput().getStatus()).isEqualTo(StatementOutputStatus.ok); }