@Test public void clientErrorOnFirstRecordsSecondRecordOk() throws Exception { SQSEvent event = new SQSEvent(); SQSEvent.SQSMessage record1 = new SQSEvent.SQSMessage(); String message = alterTableRenameJson(); String payload = "\"Message\" : \"" + message.replace("\"", "\\\"") + "\""; record1.setBody(getSnsMessage(payload)); SQSEvent.SQSMessage record2 = new SQSEvent.SQSMessage(); message = createTableJson(); payload = "\"Message\" : \"" + message.replace("\"", "\\\"") + "\""; record2.setBody(getSnsMessage(payload)); event.setRecords(Lists.newArrayList(record1, record2)); doThrow(new HiveClientException("init error")) .doNothing() .when(privilegesGrantor) .grantSelectPrivileges("some_db", "some_table1"); Response response = privilegesGrantorLambda.handleRequest(event, context); assertThat(response.getStatusCode(), is(HttpStatus.SC_INTERNAL_SERVER_ERROR)); assertThat(response.getDescription(), containsString("Failed to grant privileges")); assertThat(response.getDescription(), containsString("Privileges granted successfully")); }
@Override public Response handleRequest(SQSEvent event, Context context) { logger = context.getLogger(); List<String> failedEvents = new ArrayList<>(); List<String> successfulTableNames = new ArrayList<>(); IMetaStoreClient metaStoreClient = null; try { ThriftHiveClient thriftHiveClient = thriftHiveClientFactory .newInstance(THRIFT_CONNECTION_URI, THRIFT_CONNECTION_TIMEOUT); metaStoreClient = thriftHiveClient.getMetaStoreClient(); PrivilegesGrantor privilegesGrantor = priviligesGrantorFactory.newInstance(metaStoreClient); MessageDeserializer metaStoreEventDeserializer = defaultMessageDeserializer(); for (SQSEvent.SQSMessage record : event.getRecords()) { handleRecord(failedEvents, successfulTableNames, privilegesGrantor, metaStoreEventDeserializer, record); } } catch (HiveClientException e) { String responseMessage = String .format("Unable to establish Thrift Connection with the uri %s", THRIFT_CONNECTION_URI); logger.log(responseMessage); failedEvents.add(responseMessage); return createResponse(successfulTableNames, failedEvents); } finally { if (metaStoreClient != null) { metaStoreClient.close(); } } return createResponse(successfulTableNames, failedEvents); }
PrivilegesGrantorLambda implements RequestHandler<SQSEvent, Response> { @Override public Response handleRequest(SQSEvent event, Context context) { logger = context.getLogger(); List<String> failedEvents = new ArrayList<>(); List<String> successfulTableNames = new ArrayList<>(); IMetaStoreClient metaStoreClient = null; try { ThriftHiveClient thriftHiveClient = thriftHiveClientFactory .newInstance(THRIFT_CONNECTION_URI, THRIFT_CONNECTION_TIMEOUT); metaStoreClient = thriftHiveClient.getMetaStoreClient(); PrivilegesGrantor privilegesGrantor = priviligesGrantorFactory.newInstance(metaStoreClient); MessageDeserializer metaStoreEventDeserializer = defaultMessageDeserializer(); for (SQSEvent.SQSMessage record : event.getRecords()) { handleRecord(failedEvents, successfulTableNames, privilegesGrantor, metaStoreEventDeserializer, record); } } catch (HiveClientException e) { String responseMessage = String .format("Unable to establish Thrift Connection with the uri %s", THRIFT_CONNECTION_URI); logger.log(responseMessage); failedEvents.add(responseMessage); return createResponse(successfulTableNames, failedEvents); } finally { if (metaStoreClient != null) { metaStoreClient.close(); } } return createResponse(successfulTableNames, failedEvents); } }
PrivilegesGrantorLambda implements RequestHandler<SQSEvent, Response> { @Override public Response handleRequest(SQSEvent event, Context context) { logger = context.getLogger(); List<String> failedEvents = new ArrayList<>(); List<String> successfulTableNames = new ArrayList<>(); IMetaStoreClient metaStoreClient = null; try { ThriftHiveClient thriftHiveClient = thriftHiveClientFactory .newInstance(THRIFT_CONNECTION_URI, THRIFT_CONNECTION_TIMEOUT); metaStoreClient = thriftHiveClient.getMetaStoreClient(); PrivilegesGrantor privilegesGrantor = priviligesGrantorFactory.newInstance(metaStoreClient); MessageDeserializer metaStoreEventDeserializer = defaultMessageDeserializer(); for (SQSEvent.SQSMessage record : event.getRecords()) { handleRecord(failedEvents, successfulTableNames, privilegesGrantor, metaStoreEventDeserializer, record); } } catch (HiveClientException e) { String responseMessage = String .format("Unable to establish Thrift Connection with the uri %s", THRIFT_CONNECTION_URI); logger.log(responseMessage); failedEvents.add(responseMessage); return createResponse(successfulTableNames, failedEvents); } finally { if (metaStoreClient != null) { metaStoreClient.close(); } } return createResponse(successfulTableNames, failedEvents); } PrivilegesGrantorLambda(); @VisibleForTesting  PrivilegesGrantorLambda(
      PriviligesGrantorFactory priviligesGrantorFactory,
      ThriftHiveClientFactory thriftHiveClientFactory); }
PrivilegesGrantorLambda implements RequestHandler<SQSEvent, Response> { @Override public Response handleRequest(SQSEvent event, Context context) { logger = context.getLogger(); List<String> failedEvents = new ArrayList<>(); List<String> successfulTableNames = new ArrayList<>(); IMetaStoreClient metaStoreClient = null; try { ThriftHiveClient thriftHiveClient = thriftHiveClientFactory .newInstance(THRIFT_CONNECTION_URI, THRIFT_CONNECTION_TIMEOUT); metaStoreClient = thriftHiveClient.getMetaStoreClient(); PrivilegesGrantor privilegesGrantor = priviligesGrantorFactory.newInstance(metaStoreClient); MessageDeserializer metaStoreEventDeserializer = defaultMessageDeserializer(); for (SQSEvent.SQSMessage record : event.getRecords()) { handleRecord(failedEvents, successfulTableNames, privilegesGrantor, metaStoreEventDeserializer, record); } } catch (HiveClientException e) { String responseMessage = String .format("Unable to establish Thrift Connection with the uri %s", THRIFT_CONNECTION_URI); logger.log(responseMessage); failedEvents.add(responseMessage); return createResponse(successfulTableNames, failedEvents); } finally { if (metaStoreClient != null) { metaStoreClient.close(); } } return createResponse(successfulTableNames, failedEvents); } PrivilegesGrantorLambda(); @VisibleForTesting  PrivilegesGrantorLambda(
      PriviligesGrantorFactory priviligesGrantorFactory,
      ThriftHiveClientFactory thriftHiveClientFactory); @Override Response handleRequest(SQSEvent event, Context context); }
PrivilegesGrantorLambda implements RequestHandler<SQSEvent, Response> { @Override public Response handleRequest(SQSEvent event, Context context) { logger = context.getLogger(); List<String> failedEvents = new ArrayList<>(); List<String> successfulTableNames = new ArrayList<>(); IMetaStoreClient metaStoreClient = null; try { ThriftHiveClient thriftHiveClient = thriftHiveClientFactory .newInstance(THRIFT_CONNECTION_URI, THRIFT_CONNECTION_TIMEOUT); metaStoreClient = thriftHiveClient.getMetaStoreClient(); PrivilegesGrantor privilegesGrantor = priviligesGrantorFactory.newInstance(metaStoreClient); MessageDeserializer metaStoreEventDeserializer = defaultMessageDeserializer(); for (SQSEvent.SQSMessage record : event.getRecords()) { handleRecord(failedEvents, successfulTableNames, privilegesGrantor, metaStoreEventDeserializer, record); } } catch (HiveClientException e) { String responseMessage = String .format("Unable to establish Thrift Connection with the uri %s", THRIFT_CONNECTION_URI); logger.log(responseMessage); failedEvents.add(responseMessage); return createResponse(successfulTableNames, failedEvents); } finally { if (metaStoreClient != null) { metaStoreClient.close(); } } return createResponse(successfulTableNames, failedEvents); } PrivilegesGrantorLambda(); @VisibleForTesting  PrivilegesGrantorLambda(
      PriviligesGrantorFactory priviligesGrantorFactory,
      ThriftHiveClientFactory thriftHiveClientFactory); @Override Response handleRequest(SQSEvent event, Context context);  }