#fixed code
                              @Test
  public void testFlowQueueReadWrite() throws Exception {
    FlowQueueService service = new FlowQueueService(UTIL.getConfiguration());

    // add a couple of test flows
    Flow flow1 = createFlow(service, TEST_USER, 1);
    FlowQueueKey key1 = flow1.getQueueKey();
    Flow flow2 = createFlow(service, TEST_USER, 2);
    FlowQueueKey key2 = flow2.getQueueKey();

    // read back one flow
    Flow flow1Retrieved = service.getFlowFromQueue(key1.getCluster(), key1.getTimestamp(),
        key1.getFlowId());
    assertNotNull(flow1Retrieved);
    assertFlowEquals(key1, flow1, flow1Retrieved);

    // try reading both flows back
    List<Flow> running = service.getFlowsForStatus(TEST_CLUSTER, Flow.Status.RUNNING, 10);
    assertNotNull(running);
    assertEquals(2, running.size());

    // results should be in reverse order by timestamp
    Flow result1 = running.get(1);
    assertFlowEquals(key1, flow1, result1);
    Flow result2 = running.get(0);
    assertFlowEquals(key2, flow2, result2);

    // move both flows to successful status
    FlowQueueKey newKey1 = new FlowQueueKey(key1.getCluster(), Flow.Status.SUCCEEDED,
        key1.getTimestamp(), key1.getFlowId());
    service.moveFlow(key1, newKey1);
    FlowQueueKey newKey2 = new FlowQueueKey(key2.getCluster(), Flow.Status.SUCCEEDED,
        key2.getTimestamp(), key2.getFlowId());
    service.moveFlow(key2, newKey2);

    List<Flow> succeeded = service.getFlowsForStatus(TEST_CLUSTER, Flow.Status.SUCCEEDED, 10);
    assertNotNull(succeeded);
    assertEquals(2, succeeded.size());
    // results should still be in reverse order by timestamp
    result1 = succeeded.get(1);
    assertFlowEquals(newKey1, flow1, result1);
    result2 = succeeded.get(0);
    assertFlowEquals(newKey2, flow2, result2);

    // add flows from a second user
    Flow flow3 = createFlow(service, TEST_USER2, 3);
    FlowQueueKey key3 = flow3.getQueueKey();

    // 3rd should be the only one running
    running = service.getFlowsForStatus(TEST_CLUSTER, Flow.Status.RUNNING, 10);
    assertNotNull(running);
    assertEquals(1, running.size());
    assertFlowEquals(key3, flow3, running.get(0));

    // move flow3 to succeeded
    FlowQueueKey newKey3 = new FlowQueueKey(key3.getCluster(), Flow.Status.SUCCEEDED,
        key3.getTimestamp(), key3.getFlowId());
    service.moveFlow(key3, newKey3);

    succeeded = service.getFlowsForStatus(TEST_CLUSTER, Flow.Status.SUCCEEDED, 10);
    assertNotNull(succeeded);
    assertEquals(3, succeeded.size());
    Flow result3 = succeeded.get(0);
    assertFlowEquals(newKey3, flow3, result3);

    // test filtering by user name
    succeeded = service.getFlowsForStatus(TEST_CLUSTER, Flow.Status.SUCCEEDED, 10,
        TEST_USER2, null);
    assertNotNull(succeeded);
    assertEquals(1, succeeded.size());
    assertFlowEquals(newKey3, flow3, succeeded.get(0));

    // test pagination
    PaginatedResult<Flow> page1 = service.getPaginatedFlowsForStatus(
        TEST_CLUSTER, Flow.Status.SUCCEEDED, 1, null, null);
    List<Flow> pageValues = page1.getValues();
    assertNotNull(pageValues);
    assertNotNull(page1.getNextStartRow());
    assertEquals(1, pageValues.size());
    assertFlowEquals(newKey3, flow3, pageValues.get(0));
    // page 2
    PaginatedResult<Flow> page2 = service.getPaginatedFlowsForStatus(
        TEST_CLUSTER, Flow.Status.SUCCEEDED, 1, null, page1.getNextStartRow());
    pageValues = page2.getValues();
    assertNotNull(pageValues);
    assertNotNull(page2.getNextStartRow());
    assertEquals(1, pageValues.size());
    assertFlowEquals(newKey2, flow2, pageValues.get(0));
    // page 3
    PaginatedResult<Flow> page3 = service.getPaginatedFlowsForStatus(
        TEST_CLUSTER, Flow.Status.SUCCEEDED, 1, null, page2.getNextStartRow());
    pageValues = page3.getValues();
    assertNotNull(pageValues);
    assertNull(page3.getNextStartRow());
    assertEquals(1, pageValues.size());
    assertFlowEquals(newKey1, flow1, pageValues.get(0));
  }
#vulnerable code 
                              @Test
  public void testFlowQueueReadWrite() throws Exception {
    FlowQueueService service = new FlowQueueService(UTIL.getConfiguration());

    // add a couple of test flows
    FlowQueueKey key1 = new FlowQueueKey(TEST_CLUSTER, Flow.Status.RUNNING,
        System.currentTimeMillis(), "flow1");
    Flow flow1 = new Flow(null);
    flow1.setJobGraphJSON("{}");
    flow1.setFlowName("flow1");
    flow1.setUserName(TEST_USER);
    flow1.setProgress(10);
    service.updateFlow(key1, flow1);

    FlowQueueKey key2 = new FlowQueueKey(TEST_CLUSTER, Flow.Status.RUNNING,
        System.currentTimeMillis(), "flow2");
    Flow flow2 = new Flow(null);
    flow2.setJobGraphJSON("{}");
    flow2.setFlowName("flow2");
    flow2.setUserName(TEST_USER);
    flow2.setProgress(20);
    service.updateFlow(key2, flow2);

    // read back one flow
    Flow flow1Retrieved = service.getFlowFromQueue(key1.getCluster(), key1.getTimestamp(),
        key1.getFlowId());
    assertNotNull(flow1Retrieved);
    assertFlowEquals(key1, flow1, flow1Retrieved);

    // try reading both flows back
    List<Flow> running = service.getFlowsForStatus(TEST_CLUSTER, Flow.Status.RUNNING, 10);
    assertNotNull(running);
    assertEquals(2, running.size());

    // results should be in reverse order by timestamp
    Flow result1 = running.get(1);
    assertFlowEquals(key1, flow1, result1);
    Flow result2 = running.get(0);
    assertFlowEquals(key2, flow2, result2);

    // move both flows to successful status
    FlowQueueKey newKey1 = new FlowQueueKey(key1.getCluster(), Flow.Status.SUCCEEDED,
        key1.getTimestamp(), key1.getFlowId());
    service.moveFlow(key1, newKey1);
    FlowQueueKey newKey2 = new FlowQueueKey(key2.getCluster(), Flow.Status.SUCCEEDED,
        key2.getTimestamp(), key2.getFlowId());
    service.moveFlow(key2, newKey2);

    List<Flow> succeeded = service.getFlowsForStatus(TEST_CLUSTER, Flow.Status.SUCCEEDED, 10);
    assertNotNull(succeeded);
    assertEquals(2, succeeded.size());
    // results should still be in reverse order by timestamp
    result1 = succeeded.get(1);
    assertFlowEquals(newKey1, flow1, result1);
    result2 = succeeded.get(0);
    assertFlowEquals(newKey2, flow2, result2);
  } 
                            #location 49 
                            #vulnerability type RESOURCE_LEAK