@Test public void testAddTempResourceWithRetry() throws OdpsException { Resources resHandler = mock(Resources.class); Odps odps = mock(Odps.class); when(odps.resources()).thenReturn(resHandler); Mockito.doThrow(new OdpsException("Here is the timeout u want.")).doNothing().when(resHandler) .create(any(FileResource.class), any(InputStream.class)); MetaExplorer me = new MetaExplorerImpl(odps); String name = me.addTempResourceWithRetry(new ByteArrayInputStream(new byte[]{}), "foo", Resource.Type.FILE); Assert.assertEquals("foo_1", name); }
@Override public String addTempResourceWithRetry(InputStream in, String prefix, Resource.Type type) throws OdpsException { FileResource res = null; String extension = ""; switch (type) { case FILE: res = new FileResource(); break; case JAR: res = new JarResource(); extension = "jar"; break; default: throw new OdpsException("Unsupported resource type:" + type); } int trial = 0; String resourceName = null; res.setIsTempResource(true); while (trial <= 3) { try { resourceName = prefix + "_" + trial; if (extension != null && !extension.isEmpty()) { resourceName += "." + extension; } res.setName(resourceName); odps.resources().create(res, in); return resourceName; } catch (OdpsException e) { LOG.error("Upload resource " + resourceName + " failed:" + e.getMessage() + ", retry count=" + trial); try { Thread.sleep(60 * 1000); } catch (InterruptedException e1) { } try { in.reset(); } catch (IOException e1) { break; } } trial++; } throw new OdpsException("Upload resource failed."); }
MetaExplorerImpl implements MetaExplorer { @Override public String addTempResourceWithRetry(InputStream in, String prefix, Resource.Type type) throws OdpsException { FileResource res = null; String extension = ""; switch (type) { case FILE: res = new FileResource(); break; case JAR: res = new JarResource(); extension = "jar"; break; default: throw new OdpsException("Unsupported resource type:" + type); } int trial = 0; String resourceName = null; res.setIsTempResource(true); while (trial <= 3) { try { resourceName = prefix + "_" + trial; if (extension != null && !extension.isEmpty()) { resourceName += "." + extension; } res.setName(resourceName); odps.resources().create(res, in); return resourceName; } catch (OdpsException e) { LOG.error("Upload resource " + resourceName + " failed:" + e.getMessage() + ", retry count=" + trial); try { Thread.sleep(60 * 1000); } catch (InterruptedException e1) { } try { in.reset(); } catch (IOException e1) { break; } } trial++; } throw new OdpsException("Upload resource failed."); } }
MetaExplorerImpl implements MetaExplorer { @Override public String addTempResourceWithRetry(InputStream in, String prefix, Resource.Type type) throws OdpsException { FileResource res = null; String extension = ""; switch (type) { case FILE: res = new FileResource(); break; case JAR: res = new JarResource(); extension = "jar"; break; default: throw new OdpsException("Unsupported resource type:" + type); } int trial = 0; String resourceName = null; res.setIsTempResource(true); while (trial <= 3) { try { resourceName = prefix + "_" + trial; if (extension != null && !extension.isEmpty()) { resourceName += "." + extension; } res.setName(resourceName); odps.resources().create(res, in); return resourceName; } catch (OdpsException e) { LOG.error("Upload resource " + resourceName + " failed:" + e.getMessage() + ", retry count=" + trial); try { Thread.sleep(60 * 1000); } catch (InterruptedException e1) { } try { in.reset(); } catch (IOException e1) { break; } } trial++; } throw new OdpsException("Upload resource failed."); } MetaExplorerImpl(Odps odps); }
MetaExplorerImpl implements MetaExplorer { @Override public String addTempResourceWithRetry(InputStream in, String prefix, Resource.Type type) throws OdpsException { FileResource res = null; String extension = ""; switch (type) { case FILE: res = new FileResource(); break; case JAR: res = new JarResource(); extension = "jar"; break; default: throw new OdpsException("Unsupported resource type:" + type); } int trial = 0; String resourceName = null; res.setIsTempResource(true); while (trial <= 3) { try { resourceName = prefix + "_" + trial; if (extension != null && !extension.isEmpty()) { resourceName += "." + extension; } res.setName(resourceName); odps.resources().create(res, in); return resourceName; } catch (OdpsException e) { LOG.error("Upload resource " + resourceName + " failed:" + e.getMessage() + ", retry count=" + trial); try { Thread.sleep(60 * 1000); } catch (InterruptedException e1) { } try { in.reset(); } catch (IOException e1) { break; } } trial++; } throw new OdpsException("Upload resource failed."); } MetaExplorerImpl(Odps odps); @Override String addFileResourceWithRetry(String filePath, Resource.Type type, String padding,
                                         boolean isTempResource); @Override String addTempResourceWithRetry(InputStream in, String prefix, Resource.Type type); @Override boolean existsVolume(String projectName, String volumeName); @Override void deleteResource(String name); @Override boolean existsTable(String proj, String name); @Override boolean existsResource(String proj, String name); @Override void createFunction(List<String> resources, String project, String name, String className); @Override void deleteFunction(String project, String name); @Override Table getTable(String projectName, String tblName); @Override String getDefaultProject(); @Override Instance createInstance(Task task, int priority); String getProjectProperty(String key); }
MetaExplorerImpl implements MetaExplorer { @Override public String addTempResourceWithRetry(InputStream in, String prefix, Resource.Type type) throws OdpsException { FileResource res = null; String extension = ""; switch (type) { case FILE: res = new FileResource(); break; case JAR: res = new JarResource(); extension = "jar"; break; default: throw new OdpsException("Unsupported resource type:" + type); } int trial = 0; String resourceName = null; res.setIsTempResource(true); while (trial <= 3) { try { resourceName = prefix + "_" + trial; if (extension != null && !extension.isEmpty()) { resourceName += "." + extension; } res.setName(resourceName); odps.resources().create(res, in); return resourceName; } catch (OdpsException e) { LOG.error("Upload resource " + resourceName + " failed:" + e.getMessage() + ", retry count=" + trial); try { Thread.sleep(60 * 1000); } catch (InterruptedException e1) { } try { in.reset(); } catch (IOException e1) { break; } } trial++; } throw new OdpsException("Upload resource failed."); } MetaExplorerImpl(Odps odps); @Override String addFileResourceWithRetry(String filePath, Resource.Type type, String padding,
                                         boolean isTempResource); @Override String addTempResourceWithRetry(InputStream in, String prefix, Resource.Type type); @Override boolean existsVolume(String projectName, String volumeName); @Override void deleteResource(String name); @Override boolean existsTable(String proj, String name); @Override boolean existsResource(String proj, String name); @Override void createFunction(List<String> resources, String project, String name, String className); @Override void deleteFunction(String project, String name); @Override Table getTable(String projectName, String tblName); @Override String getDefaultProject(); @Override Instance createInstance(Task task, int priority); String getProjectProperty(String key);  }