@SuppressWarnings("unchecked") @Test public void testMultiThreadedAccessToFragments() throws Throwable { final AtomicInteger finishedCount = new AtomicInteger(); int threadCount = 100; Thread[] threads = new Thread[threadCount]; final Fragmenter fragmenter = mock(Fragmenter.class); for (int i = 0; i < threads.length; i++) { int index = i; threads[i] = new Thread(() -> { RequestParser requestParser = mock(RequestParser.class); HttpHeaders httpHeaders = mock(HttpHeaders.class); FragmenterFactory factory = mock(FragmenterFactory.class); FragmenterCacheFactory cacheFactory = mock(FragmenterCacheFactory.class); final RequestContext context = new RequestContext(); context.setTransactionId("XID-MULTI_THREADED-123456"); context.setSegmentId(index % 10); when(cacheFactory.getCache()).thenReturn(fragmentCache); when(requestParser.parseRequest(httpHeaders, RequestType.FRAGMENTER)).thenReturn(context); when(factory.getPlugin(context)).thenReturn(fragmenter); try { new FragmenterResource(requestParser, factory, cacheFactory) .getFragments(servletContext, httpHeaders); finishedCount.incrementAndGet(); } catch (Throwable e) { e.printStackTrace(); } }); threads[i].start(); } for (Thread thread : threads) { thread.join(); } verify(fragmenter, times(1)).getFragments(); assertEquals(threadCount, finishedCount.intValue()); assertEquals(1,fragmentCache.size()); fakeTicker.advanceTime(1 * 1000); fragmentCache.cleanUp(); assertEquals(1,fragmentCache.size()); fakeTicker.advanceTime(10 * 1000); fragmentCache.cleanUp(); assertEquals(0,fragmentCache.size()); }
@GET @Path("getFragments") @Produces("application/json") public Response getFragments(@Context final ServletContext servletContext, @Context final HttpHeaders headers) throws Throwable { LOG.debug("Received FRAGMENTER call"); startTime = System.currentTimeMillis(); final RequestContext context = parseRequest(headers); final String path = context.getDataSource(); final String fragmenterCacheKey = getFragmenterCacheKey(context); LOG.debug("FRAGMENTER started for path \"{}\"", path); List<Fragment> fragments; if (Utilities.isFragmenterCacheEnabled()) { try { fragments = fragmenterCacheFactory.getCache() .get(fragmenterCacheKey, new Callable<List<Fragment>>() { @Override public List<Fragment> call() throws Exception { didThreadProcessFragmentCall = true; LOG.debug("Caching fragments for transactionId={} from segmentId={} with key={}", context.getTransactionId(), context.getSegmentId(), fragmenterCacheKey); return getFragments(context); } }); } catch (UncheckedExecutionException | ExecutionException e) { if (e.getCause() != null) throw e.getCause(); throw e; } if (!didThreadProcessFragmentCall) { logFragmentStatistics(Level.DEBUG, context, fragments); } } else { LOG.debug("Fragmenter cache is disabled"); fragments = getFragments(context); } FragmentsResponse fragmentsResponse = FragmentsResponseFormatter.formatResponse(fragments, path); return Response.ok(fragmentsResponse, MediaType.APPLICATION_JSON_TYPE).build(); }
FragmenterResource extends BaseResource { @GET @Path("getFragments") @Produces("application/json") public Response getFragments(@Context final ServletContext servletContext, @Context final HttpHeaders headers) throws Throwable { LOG.debug("Received FRAGMENTER call"); startTime = System.currentTimeMillis(); final RequestContext context = parseRequest(headers); final String path = context.getDataSource(); final String fragmenterCacheKey = getFragmenterCacheKey(context); LOG.debug("FRAGMENTER started for path \"{}\"", path); List<Fragment> fragments; if (Utilities.isFragmenterCacheEnabled()) { try { fragments = fragmenterCacheFactory.getCache() .get(fragmenterCacheKey, new Callable<List<Fragment>>() { @Override public List<Fragment> call() throws Exception { didThreadProcessFragmentCall = true; LOG.debug("Caching fragments for transactionId={} from segmentId={} with key={}", context.getTransactionId(), context.getSegmentId(), fragmenterCacheKey); return getFragments(context); } }); } catch (UncheckedExecutionException | ExecutionException e) { if (e.getCause() != null) throw e.getCause(); throw e; } if (!didThreadProcessFragmentCall) { logFragmentStatistics(Level.DEBUG, context, fragments); } } else { LOG.debug("Fragmenter cache is disabled"); fragments = getFragments(context); } FragmentsResponse fragmentsResponse = FragmentsResponseFormatter.formatResponse(fragments, path); return Response.ok(fragmentsResponse, MediaType.APPLICATION_JSON_TYPE).build(); } }
FragmenterResource extends BaseResource { @GET @Path("getFragments") @Produces("application/json") public Response getFragments(@Context final ServletContext servletContext, @Context final HttpHeaders headers) throws Throwable { LOG.debug("Received FRAGMENTER call"); startTime = System.currentTimeMillis(); final RequestContext context = parseRequest(headers); final String path = context.getDataSource(); final String fragmenterCacheKey = getFragmenterCacheKey(context); LOG.debug("FRAGMENTER started for path \"{}\"", path); List<Fragment> fragments; if (Utilities.isFragmenterCacheEnabled()) { try { fragments = fragmenterCacheFactory.getCache() .get(fragmenterCacheKey, new Callable<List<Fragment>>() { @Override public List<Fragment> call() throws Exception { didThreadProcessFragmentCall = true; LOG.debug("Caching fragments for transactionId={} from segmentId={} with key={}", context.getTransactionId(), context.getSegmentId(), fragmenterCacheKey); return getFragments(context); } }); } catch (UncheckedExecutionException | ExecutionException e) { if (e.getCause() != null) throw e.getCause(); throw e; } if (!didThreadProcessFragmentCall) { logFragmentStatistics(Level.DEBUG, context, fragments); } } else { LOG.debug("Fragmenter cache is disabled"); fragments = getFragments(context); } FragmentsResponse fragmentsResponse = FragmentsResponseFormatter.formatResponse(fragments, path); return Response.ok(fragmentsResponse, MediaType.APPLICATION_JSON_TYPE).build(); } FragmenterResource();   FragmenterResource(RequestParser<HttpHeaders> parser,
                       FragmenterFactory fragmenterFactory,
                       FragmenterCacheFactory fragmenterCacheFactory); }
FragmenterResource extends BaseResource { @GET @Path("getFragments") @Produces("application/json") public Response getFragments(@Context final ServletContext servletContext, @Context final HttpHeaders headers) throws Throwable { LOG.debug("Received FRAGMENTER call"); startTime = System.currentTimeMillis(); final RequestContext context = parseRequest(headers); final String path = context.getDataSource(); final String fragmenterCacheKey = getFragmenterCacheKey(context); LOG.debug("FRAGMENTER started for path \"{}\"", path); List<Fragment> fragments; if (Utilities.isFragmenterCacheEnabled()) { try { fragments = fragmenterCacheFactory.getCache() .get(fragmenterCacheKey, new Callable<List<Fragment>>() { @Override public List<Fragment> call() throws Exception { didThreadProcessFragmentCall = true; LOG.debug("Caching fragments for transactionId={} from segmentId={} with key={}", context.getTransactionId(), context.getSegmentId(), fragmenterCacheKey); return getFragments(context); } }); } catch (UncheckedExecutionException | ExecutionException e) { if (e.getCause() != null) throw e.getCause(); throw e; } if (!didThreadProcessFragmentCall) { logFragmentStatistics(Level.DEBUG, context, fragments); } } else { LOG.debug("Fragmenter cache is disabled"); fragments = getFragments(context); } FragmentsResponse fragmentsResponse = FragmentsResponseFormatter.formatResponse(fragments, path); return Response.ok(fragmentsResponse, MediaType.APPLICATION_JSON_TYPE).build(); } FragmenterResource();   FragmenterResource(RequestParser<HttpHeaders> parser,
                       FragmenterFactory fragmenterFactory,
                       FragmenterCacheFactory fragmenterCacheFactory); @GET @Path("getFragments") @Produces("application/json") Response getFragments(@Context final ServletContext servletContext,
                                 @Context final HttpHeaders headers); @GET @Path("getFragmentsStats") @Produces("application/json") Response getFragmentsStats(@Context final ServletContext servletContext,
                                      @Context final HttpHeaders headers,
                                      @QueryParam("path") final String path); }
FragmenterResource extends BaseResource { @GET @Path("getFragments") @Produces("application/json") public Response getFragments(@Context final ServletContext servletContext, @Context final HttpHeaders headers) throws Throwable { LOG.debug("Received FRAGMENTER call"); startTime = System.currentTimeMillis(); final RequestContext context = parseRequest(headers); final String path = context.getDataSource(); final String fragmenterCacheKey = getFragmenterCacheKey(context); LOG.debug("FRAGMENTER started for path \"{}\"", path); List<Fragment> fragments; if (Utilities.isFragmenterCacheEnabled()) { try { fragments = fragmenterCacheFactory.getCache() .get(fragmenterCacheKey, new Callable<List<Fragment>>() { @Override public List<Fragment> call() throws Exception { didThreadProcessFragmentCall = true; LOG.debug("Caching fragments for transactionId={} from segmentId={} with key={}", context.getTransactionId(), context.getSegmentId(), fragmenterCacheKey); return getFragments(context); } }); } catch (UncheckedExecutionException | ExecutionException e) { if (e.getCause() != null) throw e.getCause(); throw e; } if (!didThreadProcessFragmentCall) { logFragmentStatistics(Level.DEBUG, context, fragments); } } else { LOG.debug("Fragmenter cache is disabled"); fragments = getFragments(context); } FragmentsResponse fragmentsResponse = FragmentsResponseFormatter.formatResponse(fragments, path); return Response.ok(fragmentsResponse, MediaType.APPLICATION_JSON_TYPE).build(); } FragmenterResource();   FragmenterResource(RequestParser<HttpHeaders> parser,
                       FragmenterFactory fragmenterFactory,
                       FragmenterCacheFactory fragmenterCacheFactory); @GET @Path("getFragments") @Produces("application/json") Response getFragments(@Context final ServletContext servletContext,
                                 @Context final HttpHeaders headers); @GET @Path("getFragmentsStats") @Produces("application/json") Response getFragmentsStats(@Context final ServletContext servletContext,
                                      @Context final HttpHeaders headers,
                                      @QueryParam("path") final String path);  }