@Test public void testRunWithRetryBecauseOfThrottling() throws Exception { SoftwareHeritageOriginsImporter importer = initializeImporterParams("api/origins", "somehost.com", "https", "8080", "1"); this.parameters.put(IMPORT_SOFTWARE_HERITAGE_ENDPOINT_RATELIMIT_DELAY, "1"); CloseableHttpResponse httpResponse = mock(CloseableHttpResponse.class); CloseableHttpResponse httpResponse2 = mock(CloseableHttpResponse.class); when(httpClient.execute(any(HttpHost.class),any(HttpGet.class))).thenReturn(httpResponse, httpResponse2); Gson gson = new Gson(); SoftwareHeritageOriginEntry entry = buildSoftwareHeritageOriginEntry("someUrl1"); { StatusLine statusLine = mock(StatusLine.class); when(statusLine.getStatusCode()).thenReturn(429); when(httpResponse.getStatusLine()).thenReturn(statusLine); HttpEntity httpEntity = mock(HttpEntity.class); when(httpResponse.getEntity()).thenReturn(httpEntity); String pageContents = "throttling"; InputStream pageInputStream = new ByteArrayInputStream(pageContents.getBytes()); when(httpEntity.getContentLength()).thenReturn(Long.valueOf(pageContents.length())); when(httpEntity.getContent()).thenReturn(pageInputStream); } { StatusLine statusLine = mock(StatusLine.class); when(statusLine.getStatusCode()).thenReturn(200); when(httpResponse2.getStatusLine()).thenReturn(statusLine); HttpEntity httpEntity = mock(HttpEntity.class); when(httpResponse2.getEntity()).thenReturn(httpEntity); String pageContents = gson.toJson(new SoftwareHeritageOriginEntry[] {entry}); InputStream pageInputStream = new ByteArrayInputStream(pageContents.getBytes()); when(httpEntity.getContentLength()).thenReturn(Long.valueOf(pageContents.length())); when(httpEntity.getContent()).thenReturn(pageInputStream); } importer.run(portBindings, conf, parameters); verify(originWriter, times(1)).append(originCaptor.capture()); List<SoftwareHeritageOrigin> origins = originCaptor.getAllValues(); assertEquals(1, origins.size()); assertNotNull(origins.get(0)); assertEquals(entry.getUrl(), origins.get(0).getUrl()); verifyReport(1, SoftwareHeritageOriginsImporter.COUNTER_NAME_TOTAL); }
@Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { SoftwareHeritageOriginsImporterParams params = new SoftwareHeritageOriginsImporterParams(parameters); NamedCounters counters = new NamedCounters(new String[] { COUNTER_NAME_TOTAL }); int currentCount = 0; if (StringUtils.isNotBlank(params.getShEndpointUriRoot()) && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(params.getShEndpointUriRoot())) { try (DataFileWriter<SoftwareHeritageOrigin> originsWriter = getWriter(FileSystem.get(conf), portBindings)) { long startTime = System.currentTimeMillis(); Gson gson = new Gson(); CloseableHttpClient httpclient = buildHttpClient(params.getConnectionTimeout(), params.getReadTimeout()); HttpHost target = new HttpHost(params.getShEndpointHost(), params.getShEndpointPort(), params.getShEndpointScheme()); HttpRequest getRequest = new HttpGet(buildUri(params.getShEndpointUriRoot(), params.getStartElementIndex(), params.getPageSize())); log.info("executing first page request to " + target + " with: " + getRequest.toString()); int retryCount=0; while (getRequest != null) { try (CloseableHttpResponse httpResponse = httpclient.execute(target, getRequest)) { int statusCode = httpResponse.getStatusLine().getStatusCode(); if (statusCode!=200) { if (statusCode==429) { log.warn("SH endpoint rate limit reached, delaying for " + params.getDelayMillis() + " ms, server response: " + EntityUtils.toString(httpResponse.getEntity())); Thread.sleep(params.getDelayMillis()); continue; } else { String errMessage = "got unhandled HTTP status code when accessing SH endpoint: " + statusCode + ", full status: " + httpResponse.getStatusLine() + ", server response: " + EntityUtils.toString(httpResponse.getEntity()); if (retryCount < params.getMaxRetryCount()) { retryCount++; log.error(errMessage + ", number of retries left: " + (params.getMaxRetryCount()-retryCount)); Thread.sleep(params.getDelayMillis()); continue; } else { throw new RuntimeException(errMessage); } } } else { if (retryCount > 0) { retryCount=0; } } HttpEntity entity = httpResponse.getEntity(); if (entity != null) { SoftwareHeritageOriginEntry[] entries = parsePage(EntityUtils.toString(entity), gson); if (entries != null && entries.length > 0) { for (SoftwareHeritageOriginEntry entry : entries) { originsWriter.append(convertEntry(entry)); counters.increment(COUNTER_NAME_TOTAL); currentCount++; if (currentCount % progressLogInterval == 0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis() - startTime) / 1000) + " secs"); startTime = System.currentTimeMillis(); } } } } getRequest = prepareNextRequest(httpResponse); } } log.info("total number of processed records: " + currentCount); } } else { log.warn("no endpoint URI provided, working in offline mode"); } if (counters.currentValue(COUNTER_NAME_TOTAL) == 0) { log.warn("no records imported from SH URI: " + params.getShEndpointUriRoot()); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); storeNextElementIndex(params.getStartElementIndex() + currentCount); }
SoftwareHeritageOriginsImporter implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { SoftwareHeritageOriginsImporterParams params = new SoftwareHeritageOriginsImporterParams(parameters); NamedCounters counters = new NamedCounters(new String[] { COUNTER_NAME_TOTAL }); int currentCount = 0; if (StringUtils.isNotBlank(params.getShEndpointUriRoot()) && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(params.getShEndpointUriRoot())) { try (DataFileWriter<SoftwareHeritageOrigin> originsWriter = getWriter(FileSystem.get(conf), portBindings)) { long startTime = System.currentTimeMillis(); Gson gson = new Gson(); CloseableHttpClient httpclient = buildHttpClient(params.getConnectionTimeout(), params.getReadTimeout()); HttpHost target = new HttpHost(params.getShEndpointHost(), params.getShEndpointPort(), params.getShEndpointScheme()); HttpRequest getRequest = new HttpGet(buildUri(params.getShEndpointUriRoot(), params.getStartElementIndex(), params.getPageSize())); log.info("executing first page request to " + target + " with: " + getRequest.toString()); int retryCount=0; while (getRequest != null) { try (CloseableHttpResponse httpResponse = httpclient.execute(target, getRequest)) { int statusCode = httpResponse.getStatusLine().getStatusCode(); if (statusCode!=200) { if (statusCode==429) { log.warn("SH endpoint rate limit reached, delaying for " + params.getDelayMillis() + " ms, server response: " + EntityUtils.toString(httpResponse.getEntity())); Thread.sleep(params.getDelayMillis()); continue; } else { String errMessage = "got unhandled HTTP status code when accessing SH endpoint: " + statusCode + ", full status: " + httpResponse.getStatusLine() + ", server response: " + EntityUtils.toString(httpResponse.getEntity()); if (retryCount < params.getMaxRetryCount()) { retryCount++; log.error(errMessage + ", number of retries left: " + (params.getMaxRetryCount()-retryCount)); Thread.sleep(params.getDelayMillis()); continue; } else { throw new RuntimeException(errMessage); } } } else { if (retryCount > 0) { retryCount=0; } } HttpEntity entity = httpResponse.getEntity(); if (entity != null) { SoftwareHeritageOriginEntry[] entries = parsePage(EntityUtils.toString(entity), gson); if (entries != null && entries.length > 0) { for (SoftwareHeritageOriginEntry entry : entries) { originsWriter.append(convertEntry(entry)); counters.increment(COUNTER_NAME_TOTAL); currentCount++; if (currentCount % progressLogInterval == 0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis() - startTime) / 1000) + " secs"); startTime = System.currentTimeMillis(); } } } } getRequest = prepareNextRequest(httpResponse); } } log.info("total number of processed records: " + currentCount); } } else { log.warn("no endpoint URI provided, working in offline mode"); } if (counters.currentValue(COUNTER_NAME_TOTAL) == 0) { log.warn("no records imported from SH URI: " + params.getShEndpointUriRoot()); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); storeNextElementIndex(params.getStartElementIndex() + currentCount); } }
SoftwareHeritageOriginsImporter implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { SoftwareHeritageOriginsImporterParams params = new SoftwareHeritageOriginsImporterParams(parameters); NamedCounters counters = new NamedCounters(new String[] { COUNTER_NAME_TOTAL }); int currentCount = 0; if (StringUtils.isNotBlank(params.getShEndpointUriRoot()) && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(params.getShEndpointUriRoot())) { try (DataFileWriter<SoftwareHeritageOrigin> originsWriter = getWriter(FileSystem.get(conf), portBindings)) { long startTime = System.currentTimeMillis(); Gson gson = new Gson(); CloseableHttpClient httpclient = buildHttpClient(params.getConnectionTimeout(), params.getReadTimeout()); HttpHost target = new HttpHost(params.getShEndpointHost(), params.getShEndpointPort(), params.getShEndpointScheme()); HttpRequest getRequest = new HttpGet(buildUri(params.getShEndpointUriRoot(), params.getStartElementIndex(), params.getPageSize())); log.info("executing first page request to " + target + " with: " + getRequest.toString()); int retryCount=0; while (getRequest != null) { try (CloseableHttpResponse httpResponse = httpclient.execute(target, getRequest)) { int statusCode = httpResponse.getStatusLine().getStatusCode(); if (statusCode!=200) { if (statusCode==429) { log.warn("SH endpoint rate limit reached, delaying for " + params.getDelayMillis() + " ms, server response: " + EntityUtils.toString(httpResponse.getEntity())); Thread.sleep(params.getDelayMillis()); continue; } else { String errMessage = "got unhandled HTTP status code when accessing SH endpoint: " + statusCode + ", full status: " + httpResponse.getStatusLine() + ", server response: " + EntityUtils.toString(httpResponse.getEntity()); if (retryCount < params.getMaxRetryCount()) { retryCount++; log.error(errMessage + ", number of retries left: " + (params.getMaxRetryCount()-retryCount)); Thread.sleep(params.getDelayMillis()); continue; } else { throw new RuntimeException(errMessage); } } } else { if (retryCount > 0) { retryCount=0; } } HttpEntity entity = httpResponse.getEntity(); if (entity != null) { SoftwareHeritageOriginEntry[] entries = parsePage(EntityUtils.toString(entity), gson); if (entries != null && entries.length > 0) { for (SoftwareHeritageOriginEntry entry : entries) { originsWriter.append(convertEntry(entry)); counters.increment(COUNTER_NAME_TOTAL); currentCount++; if (currentCount % progressLogInterval == 0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis() - startTime) / 1000) + " secs"); startTime = System.currentTimeMillis(); } } } } getRequest = prepareNextRequest(httpResponse); } } log.info("total number of processed records: " + currentCount); } } else { log.warn("no endpoint URI provided, working in offline mode"); } if (counters.currentValue(COUNTER_NAME_TOTAL) == 0) { log.warn("no records imported from SH URI: " + params.getShEndpointUriRoot()); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); storeNextElementIndex(params.getStartElementIndex() + currentCount); } SoftwareHeritageOriginsImporter(); }
SoftwareHeritageOriginsImporter implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { SoftwareHeritageOriginsImporterParams params = new SoftwareHeritageOriginsImporterParams(parameters); NamedCounters counters = new NamedCounters(new String[] { COUNTER_NAME_TOTAL }); int currentCount = 0; if (StringUtils.isNotBlank(params.getShEndpointUriRoot()) && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(params.getShEndpointUriRoot())) { try (DataFileWriter<SoftwareHeritageOrigin> originsWriter = getWriter(FileSystem.get(conf), portBindings)) { long startTime = System.currentTimeMillis(); Gson gson = new Gson(); CloseableHttpClient httpclient = buildHttpClient(params.getConnectionTimeout(), params.getReadTimeout()); HttpHost target = new HttpHost(params.getShEndpointHost(), params.getShEndpointPort(), params.getShEndpointScheme()); HttpRequest getRequest = new HttpGet(buildUri(params.getShEndpointUriRoot(), params.getStartElementIndex(), params.getPageSize())); log.info("executing first page request to " + target + " with: " + getRequest.toString()); int retryCount=0; while (getRequest != null) { try (CloseableHttpResponse httpResponse = httpclient.execute(target, getRequest)) { int statusCode = httpResponse.getStatusLine().getStatusCode(); if (statusCode!=200) { if (statusCode==429) { log.warn("SH endpoint rate limit reached, delaying for " + params.getDelayMillis() + " ms, server response: " + EntityUtils.toString(httpResponse.getEntity())); Thread.sleep(params.getDelayMillis()); continue; } else { String errMessage = "got unhandled HTTP status code when accessing SH endpoint: " + statusCode + ", full status: " + httpResponse.getStatusLine() + ", server response: " + EntityUtils.toString(httpResponse.getEntity()); if (retryCount < params.getMaxRetryCount()) { retryCount++; log.error(errMessage + ", number of retries left: " + (params.getMaxRetryCount()-retryCount)); Thread.sleep(params.getDelayMillis()); continue; } else { throw new RuntimeException(errMessage); } } } else { if (retryCount > 0) { retryCount=0; } } HttpEntity entity = httpResponse.getEntity(); if (entity != null) { SoftwareHeritageOriginEntry[] entries = parsePage(EntityUtils.toString(entity), gson); if (entries != null && entries.length > 0) { for (SoftwareHeritageOriginEntry entry : entries) { originsWriter.append(convertEntry(entry)); counters.increment(COUNTER_NAME_TOTAL); currentCount++; if (currentCount % progressLogInterval == 0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis() - startTime) / 1000) + " secs"); startTime = System.currentTimeMillis(); } } } } getRequest = prepareNextRequest(httpResponse); } } log.info("total number of processed records: " + currentCount); } } else { log.warn("no endpoint URI provided, working in offline mode"); } if (counters.currentValue(COUNTER_NAME_TOTAL) == 0) { log.warn("no records imported from SH URI: " + params.getShEndpointUriRoot()); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); storeNextElementIndex(params.getStartElementIndex() + currentCount); } SoftwareHeritageOriginsImporter(); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); }
SoftwareHeritageOriginsImporter implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { SoftwareHeritageOriginsImporterParams params = new SoftwareHeritageOriginsImporterParams(parameters); NamedCounters counters = new NamedCounters(new String[] { COUNTER_NAME_TOTAL }); int currentCount = 0; if (StringUtils.isNotBlank(params.getShEndpointUriRoot()) && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(params.getShEndpointUriRoot())) { try (DataFileWriter<SoftwareHeritageOrigin> originsWriter = getWriter(FileSystem.get(conf), portBindings)) { long startTime = System.currentTimeMillis(); Gson gson = new Gson(); CloseableHttpClient httpclient = buildHttpClient(params.getConnectionTimeout(), params.getReadTimeout()); HttpHost target = new HttpHost(params.getShEndpointHost(), params.getShEndpointPort(), params.getShEndpointScheme()); HttpRequest getRequest = new HttpGet(buildUri(params.getShEndpointUriRoot(), params.getStartElementIndex(), params.getPageSize())); log.info("executing first page request to " + target + " with: " + getRequest.toString()); int retryCount=0; while (getRequest != null) { try (CloseableHttpResponse httpResponse = httpclient.execute(target, getRequest)) { int statusCode = httpResponse.getStatusLine().getStatusCode(); if (statusCode!=200) { if (statusCode==429) { log.warn("SH endpoint rate limit reached, delaying for " + params.getDelayMillis() + " ms, server response: " + EntityUtils.toString(httpResponse.getEntity())); Thread.sleep(params.getDelayMillis()); continue; } else { String errMessage = "got unhandled HTTP status code when accessing SH endpoint: " + statusCode + ", full status: " + httpResponse.getStatusLine() + ", server response: " + EntityUtils.toString(httpResponse.getEntity()); if (retryCount < params.getMaxRetryCount()) { retryCount++; log.error(errMessage + ", number of retries left: " + (params.getMaxRetryCount()-retryCount)); Thread.sleep(params.getDelayMillis()); continue; } else { throw new RuntimeException(errMessage); } } } else { if (retryCount > 0) { retryCount=0; } } HttpEntity entity = httpResponse.getEntity(); if (entity != null) { SoftwareHeritageOriginEntry[] entries = parsePage(EntityUtils.toString(entity), gson); if (entries != null && entries.length > 0) { for (SoftwareHeritageOriginEntry entry : entries) { originsWriter.append(convertEntry(entry)); counters.increment(COUNTER_NAME_TOTAL); currentCount++; if (currentCount % progressLogInterval == 0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis() - startTime) / 1000) + " secs"); startTime = System.currentTimeMillis(); } } } } getRequest = prepareNextRequest(httpResponse); } } log.info("total number of processed records: " + currentCount); } } else { log.warn("no endpoint URI provided, working in offline mode"); } if (counters.currentValue(COUNTER_NAME_TOTAL) == 0) { log.warn("no records imported from SH URI: " + params.getShEndpointUriRoot()); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); storeNextElementIndex(params.getStartElementIndex() + currentCount); } SoftwareHeritageOriginsImporter(); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); static final String DEFAULT_METAFILE_NAME; }