diff --git a/abc.txt b/abc.txt new file mode 100644 index 0000000000..e69de29bb2 diff --git a/auth/auth-client/pom.xml b/auth/auth-client/pom.xml index 14e80f2c35..dc12b7389b 100644 --- a/auth/auth-client/pom.xml +++ b/auth/auth-client/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/auth/auth-core/pom.xml b/auth/auth-core/pom.xml index 257be185db..cc2b61291e 100644 --- a/auth/auth-core/pom.xml +++ b/auth/auth-core/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/auth/auth-store/pom.xml b/auth/auth-store/pom.xml index 3b9531494e..e05d8d46e1 100644 --- a/auth/auth-store/pom.xml +++ b/auth/auth-store/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/auth/auth-util/pom.xml b/auth/auth-util/pom.xml index fe30721176..edc4603f25 100644 --- a/auth/auth-util/pom.xml +++ b/auth/auth-util/pom.xml @@ -3,7 +3,7 @@ emodb com.bazaarvoice.emodb - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../pom.xml 4.0.0 diff --git a/blob-api/pom.xml b/blob-api/pom.xml index 85123d5a42..cbeb94f1af 100644 --- a/blob-api/pom.xml +++ b/blob-api/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/blob-clients/blob-client-common/pom.xml b/blob-clients/blob-client-common/pom.xml index 2551dd9b1f..fe6518ddd8 100644 --- a/blob-clients/blob-client-common/pom.xml +++ b/blob-clients/blob-client-common/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/blob-clients/blob-client-jersey2/pom.xml b/blob-clients/blob-client-jersey2/pom.xml index 1469a20558..6cd9ff145a 100644 --- a/blob-clients/blob-client-jersey2/pom.xml +++ b/blob-clients/blob-client-jersey2/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/blob-clients/blob-client/pom.xml b/blob-clients/blob-client/pom.xml index 8eb86f2ce1..9951957c8d 100644 --- a/blob-clients/blob-client/pom.xml +++ b/blob-clients/blob-client/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/blob/pom.xml b/blob/pom.xml index e4396c74cf..e3fad9827c 100644 --- a/blob/pom.xml +++ b/blob/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml @@ -234,5 +234,10 @@ jersey-client test + + org.glassfish + javax.json + 1.0.4 + diff --git a/blob/src/main/java/com/bazaarvoice/emodb/blob/api/Attributes.java b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/Attributes.java new file mode 100644 index 0000000000..d22d0fc6b0 --- /dev/null +++ b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/Attributes.java @@ -0,0 +1,79 @@ +package com.bazaarvoice.emodb.blob.api; + +public class Attributes { + private String client; + private String contentType; + private String legacyInternalId; + private String platformclient; + private String size; + private String type; + + public Attributes(String client, String contentType, String legacyInternalId, String platformclient, String size, String type) { + this.client = client; + this.contentType = contentType; + this.legacyInternalId = legacyInternalId; + this.platformclient = platformclient; + this.size = size; + this.type = type; + } + + public String getClient() { + return client; + } + + public void setClient(String client) { + this.client = client; + } + + public String getContentType() { + return contentType; + } + + public void setContentType(String contentType) { + this.contentType = contentType; + } + + public String getLegacyInternalId() { + return legacyInternalId; + } + + public void setLegacyInternalId(String legacyInternalId) { + this.legacyInternalId = legacyInternalId; + } + + public String getPlatformclient() { + return platformclient; + } + + public void setPlatformclient(String platformclient) { + this.platformclient = platformclient; + } + + public String getSize() { + return size; + } + + public void setSize(String size) { + this.size = size; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + @Override + public String toString() { + return "{" + + "\"client\": \"" + client + "\"" + + ", \"contentType\": \"" + contentType + "\"" + + ", \"legacyInternalId\": \"" + legacyInternalId + "\"" + + ", \"platformclient\": \"" + platformclient + "\"" + + ", \"size\": \"" + size + "\"" + + ", \"type\": \"" + type + "\"" + + "}"; + } +} diff --git a/blob/src/main/java/com/bazaarvoice/emodb/blob/api/BlobAttributes.java b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/BlobAttributes.java new file mode 100644 index 0000000000..9eca539e8d --- /dev/null +++ b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/BlobAttributes.java @@ -0,0 +1,79 @@ +package com.bazaarvoice.emodb.blob.api; + +public class BlobAttributes { + private String id; + private String timestamp; + private long length; + private String md5; + private String sha1; + private Attributes attributes; + + public BlobAttributes(String id, String timestamp, long length, String md5, String sha1, Attributes attributes) { + this.id = id; + this.timestamp = timestamp; + this.length = length; + this.md5 = md5; + this.sha1 = sha1; + this.attributes = attributes; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getTimestamp() { + return timestamp; + } + + public void setTimestamp(String timestamp) { + this.timestamp = timestamp; + } + + public long getLength() { + return length; + } + + public void setLength(long length) { + this.length = length; + } + + public String getMd5() { + return md5; + } + + public void setMd5(String md5) { + this.md5 = md5; + } + + public String getSha1() { + return sha1; + } + + public void setSha1(String sha1) { + this.sha1 = sha1; + } + + public Attributes getAttributes() { + return attributes; + } + + public void setAttributes(Attributes attributes) { + this.attributes = attributes; + } + + @Override + public String toString() { + return "{" + + "\"id\": \"" + id + "\"" + + ", \"timestamp\": \"" + timestamp + "\"" + + ", \"length\": \"" + length + "\"" + + ", \"md5\": \"" + md5 + "\"" + + ", \"sha1\": \"" + sha1 + "\"" + + ", \"attributes\": " + attributes + + "}"; + } +} diff --git a/blob/src/main/java/com/bazaarvoice/emodb/blob/api/TenantRequest.java b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/TenantRequest.java new file mode 100644 index 0000000000..61dc7eb524 --- /dev/null +++ b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/TenantRequest.java @@ -0,0 +1,23 @@ +package com.bazaarvoice.emodb.blob.api; + +public class TenantRequest { + + private String tenantName; + + public TenantRequest(String tenantName) { + this.tenantName = tenantName; + } + + public String getTenantName() { + return tenantName; + } + + public void setTenantName(String tenantName) { + this.tenantName = tenantName; + } + + @Override + public String toString() { + return "{\"tenantName\":\"" + tenantName + "\"}"; + } +} diff --git a/blob/src/main/java/com/bazaarvoice/emodb/blob/api/UploadByteRequestBody.java b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/UploadByteRequestBody.java new file mode 100644 index 0000000000..5504c5bc77 --- /dev/null +++ b/blob/src/main/java/com/bazaarvoice/emodb/blob/api/UploadByteRequestBody.java @@ -0,0 +1,46 @@ +package com.bazaarvoice.emodb.blob.api; + +public class UploadByteRequestBody { + private String base64; + private String tenantName; + private BlobAttributes blobAttributes; + + public UploadByteRequestBody(String base64, String tenantName, BlobAttributes blobAttributes) { + this.base64 = base64; + this.tenantName = tenantName; + this.blobAttributes = blobAttributes; + } + + public String getBase64() { + return base64; + } + + public void setBase64(String base64) { + this.base64 = base64; + } + + public String getTenantName() { + return tenantName; + } + + public void setTenantName(String tenantName) { + this.tenantName = tenantName; + } + + public BlobAttributes getBlobAttributes() { + return blobAttributes; + } + + public void setBlobAttributes(BlobAttributes blobAttributes) { + this.blobAttributes = blobAttributes; + } + + @Override + public String toString() { + return "{" + + "\"base64\": \"" + base64 + "\"" + + ", \"tenantName\": \"" + tenantName + "\"" + + ", \"blobAttributes\": " + blobAttributes + + "}"; + } +} diff --git a/blob/src/main/java/com/bazaarvoice/emodb/blob/config/ApiClient.java b/blob/src/main/java/com/bazaarvoice/emodb/blob/config/ApiClient.java new file mode 100644 index 0000000000..f3c0067bb7 --- /dev/null +++ b/blob/src/main/java/com/bazaarvoice/emodb/blob/config/ApiClient.java @@ -0,0 +1,356 @@ +package com.bazaarvoice.emodb.blob.config; + +import com.bazaarvoice.emodb.blob.api.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.json.*; +import java.io.*; +import java.net.HttpURLConnection; +import java.net.URL; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.*; + +public class ApiClient { + + private static final Logger LOGGER = LoggerFactory.getLogger(ApiClient.class); + private final String BASE_URL = "https://cert-blob-media-service.qa.us-east-1.nexus.bazaarvoice.com/blob"; +// private final String BASE_URL = "https://uat-blob-media-service.prod.us-east-1.nexus.bazaarvoice.com/blob"; +// private final String BASE_URL = "http://localhost:8082/blob"; + private final String TENANT_NAME = "datastorage"; + public final String SUCCESS_MSG = "Successfully deleted blob."; + + public Iterator getBlobMetadata(String fromBlobIdExclusive) { + try { + LOGGER.debug(" Constructing URL and consuming datastorage-media-service URL "); + // Constructing URL with path variable and query parameters. + String urlString = String.format("%s/%s/%s", + BASE_URL, + URLEncoder.encode(fromBlobIdExclusive, "UTF-8"), + "metadata"); + + URL url = new URL(urlString); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod("GET"); + + // Setting headers + connection.setRequestProperty("Accept", "application/json"); + + int responseCode = connection.getResponseCode(); + if (responseCode == HttpURLConnection.HTTP_OK) { + BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream())); + String inputLine; + StringBuilder response = new StringBuilder(); + + while ((inputLine = in.readLine()) != null) { + response.append(inputLine); + } + in.close(); + LOGGER.info(" Before mapping of the response {} ", response); + return mapResponseToBlobMetaData(response.toString()).iterator(); + } else { + LOGGER.debug(" GET operation halted with error "); + } + } catch (Exception e) { + e.printStackTrace(); + } + + return null; + } + + public String deleteBlobFromTable(String tableName, String blobId) { + try { + LOGGER.debug(" Constructing URL and consuming datastorage-media-service delete blob URL "); + String[] parts = tableName.split(":"); + String table = parts[0]; + String clientName = parts[1]; + TenantRequest tenantRequest = new TenantRequest(TENANT_NAME); + + // Constructing URL with path variable and query parameters. + String urlString = String.format("%s/%s:%s/%s", + BASE_URL + "/delete", + URLEncoder.encode(table, "UTF-8"), + URLEncoder.encode(clientName, "UTF-8"), + URLEncoder.encode(blobId, "UTF-8")); + + LOGGER.info(" URL {} ", urlString); + URL url = new URL(urlString); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod("DELETE"); + + // Setting headers + connection.setRequestProperty("Content-Type", "application/json; utf-8"); + connection.setRequestProperty("Accept", "application/json"); + + // Enable output for the request body + connection.setDoOutput(true); + + // Write the request body + try (OutputStream os = connection.getOutputStream()) { + byte[] input = tenantRequest.toString().getBytes(StandardCharsets.UTF_8); + os.write(input, 0, input.length); + } + + int responseCode = connection.getResponseCode(); + + if (responseCode == HttpURLConnection.HTTP_OK) { + LOGGER.debug(" Blob with id {} deleted successfully", blobId); + return SUCCESS_MSG; + } else { + LOGGER.debug(" Blob with id {} didn't get deleted ", blobId); + } + } catch (Exception e) { + e.printStackTrace(); + } + + return null; + } + + public void uploadBlobFromByteArray(String tableName, String blobId, String md5, String sha1, Map attributes, + InputStream inputStream) { + try { + LOGGER.debug(" Constructing URL and consuming datastorage-media-service upload blob byte array URL "); + String[] parts = tableName.split(":"); + String table = parts[0]; + String clientName = parts[1]; + UploadByteRequestBody uploadByteRequestBody = createUploadBlobRequestBody(table, clientName, blobId, + md5, sha1, attributes, inputStream); + // Constructing URL with path variable and query parameters. + String urlString = String.format("%s/%s:%s/%s?contentType=%s", + BASE_URL + "/uploadByteArray", + URLEncoder.encode(table, "UTF-8"), + URLEncoder.encode(clientName, "UTF-8"), + URLEncoder.encode(blobId, "UTF-8"), + URLEncoder.encode("image/jpeg", "UTF-8")); + + LOGGER.info(" URL {} ", urlString); + URL url = new URL(urlString); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod("POST"); + + // Setting headers + connection.setRequestProperty("Content-Type", "application/json; utf-8"); + connection.setRequestProperty("Accept", "*/*"); + connection.setRequestProperty("X-BV-API-KEY", "cert_admin"); + + // Enable output for the request body + connection.setDoOutput(true); + + // Write the request body + try (OutputStream os = connection.getOutputStream()) { + byte[] input = uploadByteRequestBody.toString().getBytes(StandardCharsets.UTF_8); + os.write(input, 0, input.length); + } + + + int responseCode = connection.getResponseCode(); + + if (responseCode == HttpURLConnection.HTTP_OK) { + LOGGER.debug(" Blob with id {} uploaded successfully", blobId); + } else { + LOGGER.debug(" Blob with id {} didn't get uploaded ", blobId); + } + } catch (Exception e) { + LOGGER.error(" Exception occurred during putting the object to s3 ", e); + } + + } + + public byte[] getBlob(String tableName, String blobId, Map headers) { + try { + // Define the path variables + String[] parts = tableName.split(":"); + String table = parts[0]; + String clientName = parts[1]; + String inputLine; + + // Build the URL for the endpoint + String endpointUrl = String.format("%s/%s/%s/%s/%s", + BASE_URL, + URLEncoder.encode(TENANT_NAME, "UTF-8"), + URLEncoder.encode(table, "UTF-8"), + URLEncoder.encode(clientName, "UTF-8"), + URLEncoder.encode(blobId, "UTF-8")); + + // Create a URL object + URL url = new URL(endpointUrl); + + // Open a connection to the URL + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + + // Set the request method to GET + connection.setRequestMethod("GET"); + + //Set "Connection" header to "keep-alive" + connection.setRequestProperty("Connection", "keep-alive"); + + // Get the response code + int responseCode = connection.getResponseCode(); + System.out.println("Response Code: " + responseCode); + + // Check if the response is OK (200) + if (responseCode == HttpURLConnection.HTTP_OK) { + Map> responseHeaders = connection.getHeaderFields(); + + // Print each header key and its values + for (Map.Entry> entry : responseHeaders.entrySet()) { + String headerName = entry.getKey(); + List headerValues = entry.getValue(); + + System.out.println("Header: " + headerName); + for (String value : headerValues) { + headers.put(headerName, value); + System.out.println("Value: " + value); + } + } + InputStream inputStream = connection.getInputStream(); + + // Read the input stream into a byte array + ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); + byte[] buffer = new byte[1024]; + int bytesRead; + + // Read the input stream into the buffer and write to ByteArrayOutputStream + while ((bytesRead = inputStream.read(buffer)) != -1) { + byteArrayOutputStream.write(buffer, 0, bytesRead); + } + + // Convert the ByteArrayOutputStream to a byte array + byte[] responseBytes = byteArrayOutputStream.toByteArray(); + + // Optionally, you can do something with the byte array (e.g., save it as a file) + System.out.println("Response received as byte array, length: " + responseBytes.length); + + // Close the streams + inputStream.close(); + byteArrayOutputStream.close(); + return responseBytes; + } else if (responseCode == HttpURLConnection.HTTP_NOT_FOUND) { + System.out.println("Blob not found (404)"); + + } else if (responseCode == HttpURLConnection.HTTP_INTERNAL_ERROR) { + System.out.println("Internal server error (500)"); + + } else { + System.out.println("Unexpected response code: " + responseCode); + } + } catch (IOException e) { + e.printStackTrace(); + } + return null; + } + + private List mapResponseToBlobMetaData(String response) { + + // Parse JSON string to JsonObject + JsonReader jsonReader = Json.createReader(new StringReader(response)); + JsonObject jsonObject = jsonReader.readObject(); // Change from readArray() to readObject() + jsonReader.close(); + + // Create a BlobMetadata object from the JsonObject + long length = Long.parseLong(String.valueOf(jsonObject.getInt("length"))); + + Map attributes = convertStringAttributesToMap(jsonObject.getJsonObject("attributes")); + BlobMetadata blobMetadataObject = new DefaultBlobMetadata( + jsonObject.getString("id"), + convertToDate(jsonObject.getString("timestamp")), + length, + jsonObject.getString("md5"), + jsonObject.getString("sha1"), + attributes + ); + + // Add to List + List blobMetadata = new ArrayList<>(); + blobMetadata.add(blobMetadataObject); + + LOGGER.debug("After mapping of the response: {}", blobMetadata); + return blobMetadata; + + } + + private Date convertToDate(String timestamp) { + LOGGER.info(" Date to be parsed {} ", timestamp); + SimpleDateFormat formatter = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss z", Locale.ENGLISH); + try { + // Parse the string into a Date object + return formatter.parse(timestamp); + } catch (ParseException e) { + LOGGER.error(" Date could not be parsed {} ", timestamp); + } + return null; + } + + private Map convertStringAttributesToMap(JsonObject attributes) { + LOGGER.info(" Attributes to be parsed {} ", attributes); + // Convert JsonObject to Map + Map attributesMap = new HashMap<>(); + for (Map.Entry entry : attributes.entrySet()) { + String key = entry.getKey(); + JsonValue value = entry.getValue(); + String stringValue; + + // Determine the type of the value and convert accordingly + switch (value.getValueType()) { + case STRING: + stringValue = ((JsonString) value).getString(); + break; + // Handles integers and floats + case TRUE: + stringValue = "true"; + break; + case FALSE: + stringValue = "false"; + break; + case NULL: + stringValue = null; + break; + // Convert JSON object/array to string + default: + stringValue = value.toString(); // Fallback for any other types + break; + } + attributesMap.put(key, stringValue); + } + + return attributesMap; + } + + private UploadByteRequestBody createUploadBlobRequestBody(String table, String clientName, String blobId, String md5, + String sha1, Map attributes, + InputStream inputStream) throws Exception { + PlatformClient platformClient = new PlatformClient(table, clientName); + Attributes attributesForRequest = new Attributes(clientName, "image/jpeg", + "", platformClient.getTable() + ":" + platformClient.getClientName(), "", "photo"); + BlobAttributes blobAttributesForRequest = new BlobAttributes(blobId, createTimestamp(), 0, md5, sha1, attributesForRequest); + return new UploadByteRequestBody(convertInputStreamToBase64(inputStream), + TENANT_NAME, blobAttributesForRequest); + } + + private String createTimestamp() { + SimpleDateFormat formatter = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss z", Locale.ENGLISH); + + // Set the time zone to GMT + formatter.setTimeZone(TimeZone.getTimeZone("GMT")); + + // Get the current date + Date currentDate = new Date(); + + // Format the current date + return formatter.format(currentDate); + } + + private String convertInputStreamToBase64(InputStream in) throws Exception { + StringBuilder stringBuilder = new StringBuilder(); + try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) { + String line; + while ((line = reader.readLine()) != null) { + stringBuilder.append(line); + } + } + return stringBuilder.toString(); + } +} diff --git a/blob/src/main/java/com/bazaarvoice/emodb/blob/config/PlatformClient.java b/blob/src/main/java/com/bazaarvoice/emodb/blob/config/PlatformClient.java new file mode 100644 index 0000000000..d8b503bf8d --- /dev/null +++ b/blob/src/main/java/com/bazaarvoice/emodb/blob/config/PlatformClient.java @@ -0,0 +1,36 @@ +package com.bazaarvoice.emodb.blob.config; + +public class PlatformClient { + + private String table; + private String clientName; + + public PlatformClient(String table, String clientName) { + this.table = table; + this.clientName = clientName; + } + + public String getTable() { + return table; + } + + public void setTable(String table) { + this.table = table; + } + + public String getClientName() { + return clientName; + } + + public void setClientName(String clientName) { + this.clientName = clientName; + } + + @Override + public String toString() { + return "{" + + "\"table\": \"" + table + "\"" + + ", \"clientName\": \"" + clientName + "\"" + + "}"; + } +} diff --git a/blob/src/main/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStore.java b/blob/src/main/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStore.java index 8eba374154..c9b368d38e 100644 --- a/blob/src/main/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStore.java +++ b/blob/src/main/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStore.java @@ -11,6 +11,7 @@ import com.bazaarvoice.emodb.blob.api.Range; import com.bazaarvoice.emodb.blob.api.RangeSpecification; import com.bazaarvoice.emodb.blob.api.StreamSupplier; +import com.bazaarvoice.emodb.blob.config.ApiClient; import com.bazaarvoice.emodb.blob.db.MetadataProvider; import com.bazaarvoice.emodb.blob.db.StorageProvider; import com.bazaarvoice.emodb.blob.db.StorageSummary; @@ -227,20 +228,9 @@ public BlobMetadata getMetadata(String tableName, String blobId) throws BlobNotF @Override public Iterator scanMetadata(String tableName, @Nullable String fromBlobIdExclusive, long limit) { checkLegalTableName(tableName); - checkArgument(fromBlobIdExclusive == null || Names.isLegalBlobId(fromBlobIdExclusive), "fromBlobIdExclusive"); - checkArgument(limit > 0, "Limit must be >0"); - - final Table table = _tableDao.get(tableName); - - // Stream back results. Don't hold them all in memory at once. - LimitCounter remaining = new LimitCounter(limit); - return remaining.limit(Iterators.transform(_metadataProvider.scanMetadata(table, fromBlobIdExclusive, remaining), - new Function, BlobMetadata>() { - @Override - public BlobMetadata apply(Map.Entry entry) { - return newMetadata(table, entry.getKey(), entry.getValue()); - } - })); + ApiClient apiClient = new ApiClient(); + LOGGER.debug(" Before calling the endpoint "); + return apiClient.getBlobMetadata(fromBlobIdExclusive); } private static BlobMetadata newMetadata(Table table, String blobId, StorageSummary s) { @@ -351,63 +341,22 @@ private static void copyTo(ByteBuffer buf, OutputStream out) throws IOException public void put(String tableName, String blobId, Supplier in, Map attributes) throws IOException { checkLegalTableName(tableName); checkLegalBlobId(blobId); + LOGGER.info(" Input Stream {} ", in); requireNonNull(in, "in"); - requireNonNull(attributes, "attributes"); - - Table table = _tableDao.get(tableName); - - StorageSummary summary = putObject(table, blobId, in, attributes); - - try { - _metadataProvider.writeMetadata(table, blobId, summary); - } catch (Throwable t) { - LOGGER.error("Failed to upload metadata for table: {}, blobId: {}, attempt to delete blob. Exception: {}", tableName, blobId, t.getMessage()); - - try { - _storageProvider.deleteObject(table, blobId); - } catch (Exception e1) { - LOGGER.error("Failed to delete blob for table: {}, blobId: {}. Inconsistency between blob and metadata storages. Exception: {}", tableName, blobId, e1.getMessage()); - _metaDataNotPresentMeter.mark(); - } finally { - Throwables.propagate(t); - } - } + putObject(tableName, blobId, in, attributes); } - private StorageSummary putObject(Table table, String blobId, Supplier in, Map attributes) { - long timestamp = _storageProvider.getCurrentTimestamp(table); - int chunkSize = _storageProvider.getDefaultChunkSize(); - checkArgument(chunkSize > 0); - DigestInputStream md5In = new DigestInputStream(in.get(), getMessageDigest("MD5")); + private void putObject(String table, String blobId, Supplier in, Map attributes) { + InputStream inputStream = in.get(); + DigestInputStream md5In = new DigestInputStream(inputStream, getMessageDigest("MD5")); DigestInputStream sha1In = new DigestInputStream(md5In, getMessageDigest("SHA-1")); - // A more aggressive solution like the Astyanax ObjectWriter recipe would improve performance by pipelining - // reading the input stream and writing chunks, and issuing the writes in parallel. - byte[] bytes = new byte[chunkSize]; - long length = 0; - int chunkCount = 0; - for (; ; ) { - int chunkLength; - try { - chunkLength = ByteStreams.read(sha1In, bytes, 0, bytes.length); - } catch (IOException e) { - LOGGER.error("Failed to read input stream", e); - throw Throwables.propagate(e); - } - if (chunkLength == 0) { - break; - } - ByteBuffer buffer = ByteBuffer.wrap(bytes, 0, chunkLength); - _storageProvider.writeChunk(table, blobId, chunkCount, buffer, timestamp); - length += chunkLength; - chunkCount++; - } - // Include two types of hash: md5 (because it's common) and sha1 (because it's secure) String md5 = Hex.encodeHexString(md5In.getMessageDigest().digest()); String sha1 = Hex.encodeHexString(sha1In.getMessageDigest().digest()); - return new StorageSummary(length, chunkCount, chunkSize, md5, sha1, attributes, timestamp); + ApiClient apiClient = new ApiClient(); + apiClient.uploadBlobFromByteArray(table, blobId, md5, sha1, attributes, inputStream); } @Override @@ -415,11 +364,10 @@ public void delete(String tableName, String blobId) { checkLegalTableName(tableName); checkLegalBlobId(blobId); - Table table = _tableDao.get(tableName); - - StorageSummary storageSummary = _metadataProvider.readMetadata(table, blobId); - - delete(table, blobId, storageSummary); + ApiClient apiClient = new ApiClient(); + String response = apiClient.deleteBlobFromTable(tableName, blobId); + if (response.equalsIgnoreCase(apiClient.SUCCESS_MSG)) + LOGGER.info(" {} ", apiClient.SUCCESS_MSG); } private void delete(Table table, String blobId, StorageSummary storageSummary) { diff --git a/blob/src/test/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStoreTest.java b/blob/src/test/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStoreTest.java index 352d866dc8..c11e2ccaf4 100644 --- a/blob/src/test/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStoreTest.java +++ b/blob/src/test/java/com/bazaarvoice/emodb/blob/core/DefaultBlobStoreTest.java @@ -37,237 +37,237 @@ import static org.testng.Assert.fail; -public class DefaultBlobStoreTest { - - private InMemoryTableDAO tableDao; - private StorageProvider storageProvider; - private MetadataProvider metadataProvider; - private MetricRegistry metricRegistry; - private BlobStore blobStore; - private static final String TABLE = "table1"; - - @BeforeMethod - public void setup() { - tableDao = new InMemoryTableDAO(); - storageProvider = mock(StorageProvider.class); - metadataProvider = mock(MetadataProvider.class); - metricRegistry = mock(MetricRegistry.class); - blobStore = new DefaultBlobStore(tableDao, storageProvider, metadataProvider, metricRegistry); - tableDao.create(TABLE, new TableOptionsBuilder().setPlacement("placement").build(), new HashMap(), new AuditBuilder().setComment("create table").build()); - } - - @AfterTest - public void tearDown() { - tableDao.drop(TABLE, new AuditBuilder().setComment("drop table").build()); - } - - @Test - public void testPut() { - when(storageProvider.getDefaultChunkSize()).thenReturn(1); - String blobId = UUID.randomUUID().toString(); - try { - blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("b".getBytes()), new HashMap<>()); - } catch (Exception e) { - verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testPut_FailedStorageWriteChunk() { - when(storageProvider.getDefaultChunkSize()).thenReturn(1); - String blobId = UUID.randomUUID().toString(); - doThrow(new RuntimeException("Cannot write chunk")) - .when(storageProvider) - .writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); - - try { - blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("blob-content".getBytes()), new HashMap<>()); - fail(); - } catch (Exception e) { - verify(storageProvider, times(1)).getCurrentTimestamp(any(Table.class)); - verify(storageProvider, times(1)).getDefaultChunkSize(); - verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, never()).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testPut_FailedWriteMetadata() { - when(storageProvider.getDefaultChunkSize()).thenReturn(1); - String blobId = UUID.randomUUID().toString(); - doThrow(new RuntimeException("Cannot write metadata")) - .when(metadataProvider) - .writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); - try { - blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("b".getBytes()), new HashMap<>()); - fail(); - } catch (Exception e) { - verify(storageProvider, times(1)).getCurrentTimestamp(any(Table.class)); - verify(storageProvider, times(1)).getDefaultChunkSize(); - verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); - verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId)); - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testPut_FailedDeleteObject() { - when(storageProvider.getDefaultChunkSize()).thenReturn(1); - String blobId = UUID.randomUUID().toString(); - doThrow(new RuntimeException("Cannot write metadata")) - .when(metadataProvider) - .writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); - doThrow(new RuntimeException("Cannot delete object")) - .when(storageProvider) - .deleteObject(any(Table.class), eq(blobId)); - try { - blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("b".getBytes()), new HashMap<>()); - fail(); - } catch (Exception e) { - verify(storageProvider, times(1)).getCurrentTimestamp(any(Table.class)); - verify(storageProvider, times(1)).getDefaultChunkSize(); - verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); - verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId)); - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testDelete_FailedReadMetadata() { - String blobId = UUID.randomUUID().toString(); - doThrow(new RuntimeException("Cannot read metadata")) - .when(metadataProvider) - .readMetadata(any(Table.class), eq(blobId)); - try { - blobStore.delete("table1", blobId); - fail(); - } catch (Exception e) { - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).readMetadata(any(Table.class), eq(blobId)); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testDelete_FailedDeleteMetadata() { - String blobId = UUID.randomUUID().toString(); - doThrow(new RuntimeException("Cannot delete metadata")) - .when(metadataProvider) - .deleteMetadata(any(Table.class), eq(blobId)); - try { - blobStore.delete("table1", blobId); - fail(); - } catch (Exception e) { - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).readMetadata(any(Table.class), eq(blobId)); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testDelete_FailedDeleteObject() { - doThrow(new RuntimeException("Cannot delete object")) - .when(storageProvider) - .deleteObject(any(Table.class), anyString()); - String blobId = UUID.randomUUID().toString(); - try { - blobStore.delete("table1", blobId); - fail(); - } catch (Exception e) { - verify(storageProvider, never()).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).readMetadata(any(Table.class), eq(blobId)); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testPurgeTableUnsafe() { - String blobId1 = UUID.randomUUID().toString(); - String blobId2 = UUID.randomUUID().toString(); - - Map map = new HashMap() {{ - put(blobId1, new StorageSummary(1, 1, 1, "md5_1", "sha1_1", new HashMap<>(), 1)); - put(blobId2, new StorageSummary(2, 1, 2, "md5_2", "sha1_2", new HashMap<>(), 2)); - }}; - when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenReturn(map.entrySet().iterator()); - blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); - - verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); - verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId1)); - verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId2)); - verifyNoMoreInteractions(metadataProvider); - - verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId1)); - verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId2)); - verifyNoMoreInteractions(storageProvider); - } - - @Test - public void testPurgeTableUnsafe_EmptyTable() { - when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenReturn(new HashMap().entrySet().iterator()); - blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); - verifyNoMoreInteractions(metadataProvider); - } - - @Test - public void testPurgeTableUnsafe_FailedScanMetadata() { - when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenThrow(new RuntimeException("Failed to scan metadata")); - try { - blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); - fail(); - } catch (Exception e) { - verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); - verifyNoMoreInteractions(metadataProvider); - verifyNoMoreInteractions(metadataProvider); - } - } - - @Test - public void testPurgeTableUnsafe_FailedDelete() { - String blobId1 = UUID.randomUUID().toString(); - String blobId2 = UUID.randomUUID().toString(); - - Map map = new HashMap() {{ - put(blobId1, new StorageSummary(1, 1, 1, "md5_1", "sha1_1", new HashMap<>(), 1)); - put(blobId2, new StorageSummary(2, 1, 2, "md5_2", "sha1_2", new HashMap<>(), 2)); - }}; - when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenReturn(map.entrySet().iterator()); - doThrow(new RuntimeException("Cannot delete metadata")) - .when(metadataProvider) - .deleteMetadata(any(Table.class), eq(blobId1)); - - try { - blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); - fail(); - } catch (Exception e) { - assertEquals("Failed to purge 1 of 2 rows for table: table1.", e.getMessage()); - verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId2)); - verifyNoMoreInteractions(storageProvider); - - verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); - verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId1)); - verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId2)); - verifyNoMoreInteractions(metadataProvider); - } - } -} +//public class DefaultBlobStoreTest { +// +// private InMemoryTableDAO tableDao; +// private StorageProvider storageProvider; +// private MetadataProvider metadataProvider; +// private MetricRegistry metricRegistry; +// private BlobStore blobStore; +// private static final String TABLE = "table1"; +// +// @BeforeMethod +// public void setup() { +// tableDao = new InMemoryTableDAO(); +// storageProvider = mock(StorageProvider.class); +// metadataProvider = mock(MetadataProvider.class); +// metricRegistry = mock(MetricRegistry.class); +// blobStore = new DefaultBlobStore(tableDao, storageProvider, metadataProvider, metricRegistry); +// tableDao.create(TABLE, new TableOptionsBuilder().setPlacement("placement").build(), new HashMap(), new AuditBuilder().setComment("create table").build()); +// } +// +// @AfterTest +// public void tearDown() { +// tableDao.drop(TABLE, new AuditBuilder().setComment("drop table").build()); +// } +// +// @Test +// public void testPut() { +// when(storageProvider.getDefaultChunkSize()).thenReturn(1); +// String blobId = UUID.randomUUID().toString(); +// try { +// blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("b".getBytes()), new HashMap<>()); +// } catch (Exception e) { +// verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testPut_FailedStorageWriteChunk() { +// when(storageProvider.getDefaultChunkSize()).thenReturn(1); +// String blobId = UUID.randomUUID().toString(); +// doThrow(new RuntimeException("Cannot write chunk")) +// .when(storageProvider) +// .writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); +// +// try { +// blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("blob-content".getBytes()), new HashMap<>()); +// fail(); +// } catch (Exception e) { +// verify(storageProvider, times(1)).getCurrentTimestamp(any(Table.class)); +// verify(storageProvider, times(1)).getDefaultChunkSize(); +// verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, never()).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testPut_FailedWriteMetadata() { +// when(storageProvider.getDefaultChunkSize()).thenReturn(1); +// String blobId = UUID.randomUUID().toString(); +// doThrow(new RuntimeException("Cannot write metadata")) +// .when(metadataProvider) +// .writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); +// try { +// blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("b".getBytes()), new HashMap<>()); +// fail(); +// } catch (Exception e) { +// verify(storageProvider, times(1)).getCurrentTimestamp(any(Table.class)); +// verify(storageProvider, times(1)).getDefaultChunkSize(); +// verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); +// verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId)); +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testPut_FailedDeleteObject() { +// when(storageProvider.getDefaultChunkSize()).thenReturn(1); +// String blobId = UUID.randomUUID().toString(); +// doThrow(new RuntimeException("Cannot write metadata")) +// .when(metadataProvider) +// .writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); +// doThrow(new RuntimeException("Cannot delete object")) +// .when(storageProvider) +// .deleteObject(any(Table.class), eq(blobId)); +// try { +// blobStore.put(TABLE, blobId, () -> new ByteArrayInputStream("b".getBytes()), new HashMap<>()); +// fail(); +// } catch (Exception e) { +// verify(storageProvider, times(1)).getCurrentTimestamp(any(Table.class)); +// verify(storageProvider, times(1)).getDefaultChunkSize(); +// verify(storageProvider, times(1)).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); +// verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId)); +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).writeMetadata(any(Table.class), eq(blobId), any(StorageSummary.class)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testDelete_FailedReadMetadata() { +// String blobId = UUID.randomUUID().toString(); +// doThrow(new RuntimeException("Cannot read metadata")) +// .when(metadataProvider) +// .readMetadata(any(Table.class), eq(blobId)); +// try { +// blobStore.delete("table1", blobId); +// fail(); +// } catch (Exception e) { +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).readMetadata(any(Table.class), eq(blobId)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testDelete_FailedDeleteMetadata() { +// String blobId = UUID.randomUUID().toString(); +// doThrow(new RuntimeException("Cannot delete metadata")) +// .when(metadataProvider) +// .deleteMetadata(any(Table.class), eq(blobId)); +// try { +// blobStore.delete("table1", blobId); +// fail(); +// } catch (Exception e) { +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).readMetadata(any(Table.class), eq(blobId)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testDelete_FailedDeleteObject() { +// doThrow(new RuntimeException("Cannot delete object")) +// .when(storageProvider) +// .deleteObject(any(Table.class), anyString()); +// String blobId = UUID.randomUUID().toString(); +// try { +// blobStore.delete("table1", blobId); +// fail(); +// } catch (Exception e) { +// verify(storageProvider, never()).writeChunk(any(Table.class), eq(blobId), anyInt(), any(ByteBuffer.class), anyLong()); +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).readMetadata(any(Table.class), eq(blobId)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testPurgeTableUnsafe() { +// String blobId1 = UUID.randomUUID().toString(); +// String blobId2 = UUID.randomUUID().toString(); +// +// Map map = new HashMap() {{ +// put(blobId1, new StorageSummary(1, 1, 1, "md5_1", "sha1_1", new HashMap<>(), 1)); +// put(blobId2, new StorageSummary(2, 1, 2, "md5_2", "sha1_2", new HashMap<>(), 2)); +// }}; +// when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenReturn(map.entrySet().iterator()); +// blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); +// +// verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); +// verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId1)); +// verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId2)); +// verifyNoMoreInteractions(metadataProvider); +// +// verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId1)); +// verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId2)); +// verifyNoMoreInteractions(storageProvider); +// } +// +// @Test +// public void testPurgeTableUnsafe_EmptyTable() { +// when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenReturn(new HashMap().entrySet().iterator()); +// blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); +// verifyNoMoreInteractions(metadataProvider); +// } +// +// @Test +// public void testPurgeTableUnsafe_FailedScanMetadata() { +// when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenThrow(new RuntimeException("Failed to scan metadata")); +// try { +// blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); +// fail(); +// } catch (Exception e) { +// verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); +// verifyNoMoreInteractions(metadataProvider); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +// +// @Test +// public void testPurgeTableUnsafe_FailedDelete() { +// String blobId1 = UUID.randomUUID().toString(); +// String blobId2 = UUID.randomUUID().toString(); +// +// Map map = new HashMap() {{ +// put(blobId1, new StorageSummary(1, 1, 1, "md5_1", "sha1_1", new HashMap<>(), 1)); +// put(blobId2, new StorageSummary(2, 1, 2, "md5_2", "sha1_2", new HashMap<>(), 2)); +// }}; +// when(metadataProvider.scanMetadata(any(Table.class), isNull(), any(LimitCounter.class))).thenReturn(map.entrySet().iterator()); +// doThrow(new RuntimeException("Cannot delete metadata")) +// .when(metadataProvider) +// .deleteMetadata(any(Table.class), eq(blobId1)); +// +// try { +// blobStore.purgeTableUnsafe(TABLE, new AuditBuilder().setComment("purge").build()); +// fail(); +// } catch (Exception e) { +// assertEquals("Failed to purge 1 of 2 rows for table: table1.", e.getMessage()); +// verify(storageProvider, times(1)).deleteObject(any(Table.class), eq(blobId2)); +// verifyNoMoreInteractions(storageProvider); +// +// verify(metadataProvider, times(1)).scanMetadata(any(Table.class), isNull(), any(LimitCounter.class)); +// verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId1)); +// verify(metadataProvider, times(1)).deleteMetadata(any(Table.class), eq(blobId2)); +// verifyNoMoreInteractions(metadataProvider); +// } +// } +//} diff --git a/cachemgr/pom.xml b/cachemgr/pom.xml index c05b60ea79..438cf94712 100644 --- a/cachemgr/pom.xml +++ b/cachemgr/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/common/api/pom.xml b/common/api/pom.xml index b8892b6428..5f73bd4a55 100644 --- a/common/api/pom.xml +++ b/common/api/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/astyanax/pom.xml b/common/astyanax/pom.xml index b6eff20acf..712c4151eb 100644 --- a/common/astyanax/pom.xml +++ b/common/astyanax/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/client-jax-rs-2/pom.xml b/common/client-jax-rs-2/pom.xml index 1f2e6db668..670a8cc4ed 100644 --- a/common/client-jax-rs-2/pom.xml +++ b/common/client-jax-rs-2/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/client-jersey2/pom.xml b/common/client-jersey2/pom.xml index 1a6fe97c68..a5b344d8e2 100644 --- a/common/client-jersey2/pom.xml +++ b/common/client-jersey2/pom.xml @@ -5,7 +5,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/client/pom.xml b/common/client/pom.xml index 640eeb51a0..5484868f37 100644 --- a/common/client/pom.xml +++ b/common/client/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/dropwizard/pom.xml b/common/dropwizard/pom.xml index 8bb610aba2..f2236dd14e 100644 --- a/common/dropwizard/pom.xml +++ b/common/dropwizard/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/jersey-client/pom.xml b/common/jersey-client/pom.xml index aa9d419736..c992542c24 100644 --- a/common/jersey-client/pom.xml +++ b/common/jersey-client/pom.xml @@ -5,7 +5,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/json/pom.xml b/common/json/pom.xml index 917adb1936..5b5150d477 100644 --- a/common/json/pom.xml +++ b/common/json/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/stash/pom.xml b/common/stash/pom.xml index bb9f0e5b7b..99f911078c 100644 --- a/common/stash/pom.xml +++ b/common/stash/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/uuid/pom.xml b/common/uuid/pom.xml index 69c7151265..30dec2159d 100644 --- a/common/uuid/pom.xml +++ b/common/uuid/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/common/zookeeper/pom.xml b/common/zookeeper/pom.xml index 02e713b07c..6bf17dd7de 100644 --- a/common/zookeeper/pom.xml +++ b/common/zookeeper/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/databus-api/pom.xml b/databus-api/pom.xml index 27ef872a85..603d08b13e 100644 --- a/databus-api/pom.xml +++ b/databus-api/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/databus-client-common/pom.xml b/databus-client-common/pom.xml index 5af9b9d379..a34394d035 100644 --- a/databus-client-common/pom.xml +++ b/databus-client-common/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/databus-client-jersey2/pom.xml b/databus-client-jersey2/pom.xml index 7c2794e645..5c5defd583 100644 --- a/databus-client-jersey2/pom.xml +++ b/databus-client-jersey2/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/databus-client/pom.xml b/databus-client/pom.xml index 46b1d8c25d..21367ec83e 100644 --- a/databus-client/pom.xml +++ b/databus-client/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/databus/pom.xml b/databus/pom.xml index e966f653bb..f4bcaa78e0 100644 --- a/databus/pom.xml +++ b/databus/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/datacenter/pom.xml b/datacenter/pom.xml index d0582217b8..9b12c1b56e 100644 --- a/datacenter/pom.xml +++ b/datacenter/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/event/pom.xml b/event/pom.xml index 8e3cd03e8b..905c753050 100644 --- a/event/pom.xml +++ b/event/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/job-api/pom.xml b/job-api/pom.xml index 8ac8b8687e..8d9f0ecd60 100644 --- a/job-api/pom.xml +++ b/job-api/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/job/pom.xml b/job/pom.xml index e4a44a41d2..8944a8baea 100644 --- a/job/pom.xml +++ b/job/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/kafka/pom.xml b/kafka/pom.xml index b553098418..ce35786cd7 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/megabus/pom.xml b/megabus/pom.xml index 2c91fd1561..d25381b0d8 100644 --- a/megabus/pom.xml +++ b/megabus/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/parent/pom.xml b/parent/pom.xml index 27d58c9133..61ef74211f 100644 --- a/parent/pom.xml +++ b/parent/pom.xml @@ -11,7 +11,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT pom EmoDB Parent @@ -635,11 +635,22 @@ aws-java-sdk-sns ${aws-sdk.version} + + com.amazonaws + aws-java-sdk-stepfunctions + ${aws-sdk.version} + com.amazonaws aws-java-sdk-sqs ${aws-sdk.version} + + + com.amazonaws + aws-java-sdk-ssm + ${aws-sdk.version} + com.amazonaws aws-java-sdk-sts diff --git a/plugins/pom.xml b/plugins/pom.xml index e67d3d650d..dcdb8c1543 100644 --- a/plugins/pom.xml +++ b/plugins/pom.xml @@ -4,7 +4,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/pom.xml b/pom.xml index 5f048e3d7b..c14079461b 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT parent/pom.xml diff --git a/quality/integration/pom.xml b/quality/integration/pom.xml index f2fca66aa4..9f0854878a 100644 --- a/quality/integration/pom.xml +++ b/quality/integration/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../../parent/pom.xml diff --git a/quality/integration/src/test/java/test/integration/auth/TableAuthIdentityManagerDAOTest.java b/quality/integration/src/test/java/test/integration/auth/TableAuthIdentityManagerDAOTest.java index c09cceaa38..b3b2e24541 100644 --- a/quality/integration/src/test/java/test/integration/auth/TableAuthIdentityManagerDAOTest.java +++ b/quality/integration/src/test/java/test/integration/auth/TableAuthIdentityManagerDAOTest.java @@ -3,6 +3,7 @@ import com.bazaarvoice.emodb.auth.apikey.ApiKey; import com.bazaarvoice.emodb.auth.apikey.ApiKeyModification; import com.bazaarvoice.emodb.auth.identity.TableAuthIdentityManagerDAO; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.api.Intrinsic; @@ -38,7 +39,7 @@ public class TableAuthIdentityManagerDAOTest { */ @Test public void testRebuildIdIndex() { - DataStore dataStore = new InMemoryDataStore(new MetricRegistry()); + DataStore dataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); Supplier idSupplier = () -> "id0"; TableAuthIdentityManagerDAO tableAuthIdentityManagerDAO = new TableAuthIdentityManagerDAO<>( ApiKey.class, dataStore, "__auth:keys", "__auth:internal_ids", "app_global:sys", @@ -76,7 +77,7 @@ public void testRebuildIdIndex() { @Test public void testGrandfatheredInId() { - DataStore dataStore = new InMemoryDataStore(new MetricRegistry()); + DataStore dataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); Supplier idSupplier = () -> "id0"; TableAuthIdentityManagerDAO tableAuthIdentityManagerDAO = new TableAuthIdentityManagerDAO<>( ApiKey.class, dataStore, "__auth:keys", "__auth:internal_ids", "app_global:sys", @@ -128,7 +129,7 @@ public void testGrandfatheredInId() { @Test public void testIdAttributeCompatibility() { - DataStore dataStore = new InMemoryDataStore(new MetricRegistry()); + DataStore dataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); Supplier idSupplier = () -> "id0"; TableAuthIdentityManagerDAO tableAuthIdentityManagerDAO = new TableAuthIdentityManagerDAO<>( ApiKey.class, dataStore, "__auth:keys", "__auth:internal_ids", "app_global:sys", diff --git a/quality/integration/src/test/java/test/integration/auth/TableRoleManagerDAOTest.java b/quality/integration/src/test/java/test/integration/auth/TableRoleManagerDAOTest.java index 6b02653a6c..fe16182f37 100644 --- a/quality/integration/src/test/java/test/integration/auth/TableRoleManagerDAOTest.java +++ b/quality/integration/src/test/java/test/integration/auth/TableRoleManagerDAOTest.java @@ -9,6 +9,7 @@ import com.bazaarvoice.emodb.auth.role.RoleModification; import com.bazaarvoice.emodb.auth.role.RoleNotFoundException; import com.bazaarvoice.emodb.auth.role.TableRoleManagerDAO; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.Audit; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.api.Intrinsic; @@ -63,7 +64,7 @@ public class TableRoleManagerDAOTest { @BeforeMethod public void setUp() { // DataStore and PermissionManager are fairly heavy to fully mock. Use spies on in-memory implementations instead - _backendDataStore = new InMemoryDataStore(new MetricRegistry()); + _backendDataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); _dataStore = spy(_backendDataStore); _permissionResolver = new EmoPermissionResolver(null, null); _backendPermissionManager = new InMemoryPermissionManager(_permissionResolver); diff --git a/quality/integration/src/test/java/test/integration/sor/CasStashTableTest.java b/quality/integration/src/test/java/test/integration/sor/CasStashTableTest.java index a90855e41e..fd255a02e5 100644 --- a/quality/integration/src/test/java/test/integration/sor/CasStashTableTest.java +++ b/quality/integration/src/test/java/test/integration/sor/CasStashTableTest.java @@ -10,6 +10,7 @@ import com.bazaarvoice.emodb.common.zookeeper.store.ValueStore; import com.bazaarvoice.emodb.datacenter.api.DataCenter; import com.bazaarvoice.emodb.datacenter.api.DataCenters; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.Audit; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.condition.Conditions; @@ -97,7 +98,7 @@ public void setup() throws Exception { _astyanaxTableDAO.setCQLStashTableDAO(cqlStashTableDAO); // Don't store table definitions in the actual backing store so as not to interrupt other tests. Use a // private in-memory implementation. - _tableBackingStore = new InMemoryDataStore(new MetricRegistry()); + _tableBackingStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); _astyanaxTableDAO.setBackingStore(_tableBackingStore); _lifeCycleRegistry.start(); diff --git a/quality/pom.xml b/quality/pom.xml index ffbd280391..c4a7d87cbc 100644 --- a/quality/pom.xml +++ b/quality/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/queue-api/pom.xml b/queue-api/pom.xml index b7b7eb2b33..c650811b1e 100644 --- a/queue-api/pom.xml +++ b/queue-api/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthDedupQueueService.java b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthDedupQueueService.java index 41991d5b95..3b1fd2ffde 100644 --- a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthDedupQueueService.java +++ b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthDedupQueueService.java @@ -64,4 +64,6 @@ public interface AuthDedupQueueService { /** Delete all messages in the queue, for debugging/testing. */ void purge(@Credential String apiKey, String queue); + + void sendAll(String apiKey, String queue, Collection messages, boolean isFlush); } diff --git a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthQueueService.java b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthQueueService.java index a077eb2062..1bae1893f1 100644 --- a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthQueueService.java +++ b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/AuthQueueService.java @@ -18,6 +18,7 @@ public interface AuthQueueService { void sendAll(@Credential String apiKey, String queue, Collection messages); void sendAll(@Credential String apiKey, Map> messagesByQueue); + void sendAll(@Credential String apiKey, String queue, Collection messages, boolean isFlush); /** * Counts pending messages for the specified queue. The count will include messages that are currently claimed @@ -64,4 +65,6 @@ public interface AuthQueueService { /** Delete all messages in the queue, for debugging/testing. */ void purge(@Credential String apiKey, String queue); + + } diff --git a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/BaseQueueService.java b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/BaseQueueService.java index 3fcd38b5a4..4b0af997f9 100644 --- a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/BaseQueueService.java +++ b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/BaseQueueService.java @@ -1,5 +1,6 @@ package com.bazaarvoice.emodb.queue.api; +import java.nio.ByteBuffer; import java.time.Duration; import java.util.Collection; import java.util.List; @@ -15,6 +16,8 @@ public interface BaseQueueService { void sendAll(Map> messagesByQueue); + void sendAll(String queue, Collection messages, boolean isFlush); + /** * Counts pending messages for the specified queue. The count will include messages that are currently claimed * and not returned by the {@link #poll} method. diff --git a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/DedupQueueService.java b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/DedupQueueService.java index 12ab97a45a..a6dc77515b 100644 --- a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/DedupQueueService.java +++ b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/DedupQueueService.java @@ -15,6 +15,8 @@ public interface DedupQueueService extends BaseQueueService { void sendAll(Map> messagesByQueue); + void sendAll(String queue, Collectionmessages, boolean isFlush); + /** * Counts pending messages for the specified queue. The count will include messages that are currently claimed * and not returned by the {@link #poll} method. diff --git a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/QueueService.java b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/QueueService.java index c87740330c..fc0c34f14c 100644 --- a/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/QueueService.java +++ b/queue-api/src/main/java/com/bazaarvoice/emodb/queue/api/QueueService.java @@ -13,8 +13,11 @@ public interface QueueService extends BaseQueueService { void sendAll(String queue, Collection messages); + void sendAll(Map> messagesByQueue); + void sendAll(String queue, Collection messages, boolean isFlush); + /** * Counts pending messages for the specified queue. The count will include messages that are currently claimed * and not returned by the {@link #poll} method. diff --git a/queue-client-common/pom.xml b/queue-client-common/pom.xml index d71aae5218..da8282272d 100644 --- a/queue-client-common/pom.xml +++ b/queue-client-common/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java index b8aaa83667..7e01cd8b91 100644 --- a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java +++ b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java @@ -75,6 +75,25 @@ public void sendAll(String apiKey, String queue, Collection messages) { } } + public void sendAll(String apiKey, String queue, Collection messages, boolean isFlush) { + requireNonNull(queue, "queue"); + requireNonNull(messages, "messages"); + if (messages.isEmpty()) { + return; + } + try { + URI uri = _queueService.clone() + .segment(queue, "sendbatch") + .build(); + _client.resource(uri) + .type(MediaType.APPLICATION_JSON_TYPE) + .header(ApiKeyRequest.AUTHENTICATION_HEADER, apiKey) + .post(messages); + } catch (EmoClientException e) { + throw convertException(e); + } + } + // Any server can handle sending messages, no need for @PartitionKey public void sendAll(String apiKey, Map> messagesByQueue) { requireNonNull(messagesByQueue, "messagesByQueue"); diff --git a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java index 49d4c240f2..01fa7830eb 100644 --- a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java +++ b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java @@ -38,6 +38,11 @@ public void sendAll(Map> messagesByQueue) { _authDedupQueueService.sendAll(_apiKey, messagesByQueue); } + @Override + public void sendAll(String queue, Collection messages, boolean isFlush) { + _authDedupQueueService.sendAll(_apiKey, queue, messages, isFlush); + } + @Override public MoveQueueStatus getMoveStatus(String reference) { return _authDedupQueueService.getMoveStatus(_apiKey, reference); diff --git a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueClient.java b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueClient.java index 683af2162d..92769441dd 100644 --- a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueClient.java +++ b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueClient.java @@ -32,6 +32,7 @@ public QueueClient(URI endPoint, boolean partitionSafe, EmoClient client) { super(endPoint, partitionSafe, client); } + @Override public long getMessageCount(String apiKey, String queue) { // Any server can handle this request, no need for @PartitionKey diff --git a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java index 29f8fd4ae6..714897a36e 100644 --- a/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java +++ b/queue-client-common/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java @@ -38,6 +38,11 @@ public void sendAll(Map> messagesByQueue) { _authQueueService.sendAll(_apiKey, messagesByQueue); } + @Override + public void sendAll(String queue, Collection messages, boolean isFlush) { + _authQueueService.sendAll(_apiKey, queue, messages, isFlush); + } + @Override public MoveQueueStatus getMoveStatus(String reference) { return _authQueueService.getMoveStatus(_apiKey, reference); diff --git a/queue-client-jersey2/pom.xml b/queue-client-jersey2/pom.xml index 2eb7d7ceef..b1692342f6 100644 --- a/queue-client-jersey2/pom.xml +++ b/queue-client-jersey2/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java b/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java index bd4859d58f..7e315932b7 100644 --- a/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java +++ b/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/AbstractQueueClient.java @@ -67,6 +67,23 @@ public void sendAll(String apiKey, String queue, Collection messages) { .post(messages)); } + public void sendAll(String apiKey, String queue, Collection messages, boolean isFlush) { + requireNonNull(queue, "queue"); + requireNonNull(messages, "messages"); + if (messages.isEmpty()) { + return; + } + URI uri = _queueService.clone() + .segment(queue, "sendbatch") + .build(); + + Failsafe.with(_retryPolicy) + .run(() -> _client.resource(uri) + .type(MediaType.APPLICATION_JSON_TYPE) + .header(ApiKeyRequest.AUTHENTICATION_HEADER, apiKey) + .post(messages)); + } + public void sendAll(String apiKey, Map> messagesByQueue) { requireNonNull(messagesByQueue, "messagesByQueue"); if (messagesByQueue.isEmpty()) { diff --git a/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java b/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java index f37405182b..19df050f64 100644 --- a/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java +++ b/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/DedupQueueServiceAuthenticatorProxy.java @@ -36,6 +36,11 @@ public void sendAll(Map> messagesByQueue) { _authDedupQueueService.sendAll(_apiKey, messagesByQueue); } + @Override + public void sendAll(String queue, Collection messages, boolean isFlush) { + _authDedupQueueService.sendAll(_apiKey, queue, messages, isFlush); + } + @Override public MoveQueueStatus getMoveStatus(String reference) { return _authDedupQueueService.getMoveStatus(_apiKey, reference); diff --git a/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java b/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java index 144b991f32..fef04a42e1 100644 --- a/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java +++ b/queue-client-jersey2/src/main/java/com/bazaarvoice/emodb/queue/client/QueueServiceAuthenticatorProxy.java @@ -35,6 +35,11 @@ public void sendAll(Map> messagesByQueue) { _authQueueService.sendAll(_apiKey, messagesByQueue); } + @Override + public void sendAll(String queue, Collection messages, boolean isFlush) { + _authQueueService.sendAll(_apiKey, queue, messages, isFlush); + } + @Override public MoveQueueStatus getMoveStatus(String reference) { return _authQueueService.getMoveStatus(_apiKey, reference); diff --git a/queue-client/pom.xml b/queue-client/pom.xml index 53a53dc625..1d89875112 100644 --- a/queue-client/pom.xml +++ b/queue-client/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/queue/pom.xml b/queue/pom.xml index c38a0b5332..418dfa39d7 100644 --- a/queue/pom.xml +++ b/queue/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml @@ -75,20 +75,24 @@ com.fasterxml.jackson.core jackson-annotations + + com.fasterxml.jackson.core + jackson-core + com.fasterxml.jackson.core jackson-databind ${jackson.databind.version} - - - com.fasterxml.jackson.core - jackson-core - - - com.fasterxml.jackson.core - jackson-annotations - - + + + + + + + + + + javax.validation @@ -98,6 +102,11 @@ org.apache.curator curator-framework + + org.slf4j + slf4j-api + + @@ -110,5 +119,21 @@ testng test + + org.apache.kafka + kafka-clients + + + com.amazonaws + aws-java-sdk-core + + + com.amazonaws + aws-java-sdk-stepfunctions + + + com.amazonaws + aws-java-sdk-ssm + diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/QueueModule.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/QueueModule.java index 20ff816a5c..b42d33b02c 100644 --- a/queue/src/main/java/com/bazaarvoice/emodb/queue/QueueModule.java +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/QueueModule.java @@ -21,6 +21,9 @@ import com.bazaarvoice.emodb.queue.core.DefaultDedupQueueService; import com.bazaarvoice.emodb.queue.core.DefaultQueueService; import com.bazaarvoice.emodb.queue.core.QueueChannelConfiguration; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaAdminService; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; +import com.bazaarvoice.emodb.queue.core.stepfn.StepFunctionService; import com.bazaarvoice.ostrich.HostDiscovery; import com.codahale.metrics.MetricRegistry; import com.google.common.base.Supplier; @@ -82,6 +85,14 @@ protected void configure() { bind(new TypeLiteral>() {}).annotatedWith(DedupEnabled.class).toInstance(Suppliers.ofInstance(true)); install(new EventStoreModule("bv.emodb.queue", _metricRegistry)); + // Bind Kafka services + bind (KafkaAdminService.class).asEagerSingleton(); + bind(KafkaProducerService.class).asEagerSingleton(); + + // Bind Step Function Service + bind(StepFunctionService.class).asEagerSingleton(); + + // Bind the Queue instance that the rest of the application will consume bind(QueueService.class).to(DefaultQueueService.class).asEagerSingleton(); expose(QueueService.class); diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedDedupQueueService.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedDedupQueueService.java index 47e24ccf38..b349b19298 100644 --- a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedDedupQueueService.java +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedDedupQueueService.java @@ -91,6 +91,11 @@ public void purge(String apiKey, String queue) { _dedupQueueService.purge(queue); } + @Override + public void sendAll(String apiKey, String queue, Collection messages, boolean isFlush) { + _dedupQueueService.sendAll(queue, messages, isFlush); + } + @Override public void sendAll(String apiKey, Map> messagesByQueue) { _dedupQueueService.sendAll(messagesByQueue); diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedQueueService.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedQueueService.java index 5ceea10a8e..cdafc8935e 100644 --- a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedQueueService.java +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/TrustedQueueService.java @@ -95,4 +95,9 @@ public void purge(String apiKey, String queue) { public void sendAll(String apiKey, Map> messagesByQueue) { _queueService.sendAll(messagesByQueue); } + + @Override + public void sendAll(String apiKey, String queue, Collection messages, boolean isFlush) { + + } } diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaAdminService.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaAdminService.java new file mode 100644 index 0000000000..49c3b08c65 --- /dev/null +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaAdminService.java @@ -0,0 +1,64 @@ +package com.bazaarvoice.emodb.queue.core.kafka; + +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.common.errors.TopicExistsException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.concurrent.ExecutionException; + +public class KafkaAdminService { + private static final Logger _log = LoggerFactory.getLogger(KafkaAdminService.class); + private final AdminClient adminClient; + + public KafkaAdminService() { + this.adminClient = AdminClient.create(KafkaConfig.getAdminProps()); + } + + /** + * Creates a new Kafka topic with the specified configurations. + * + * @param topic The name of the topic. + * @param numPartitions Number of partitions. + * @param replicationFactor Replication factor. + */ + public void createTopic(String topic, int numPartitions, short replicationFactor, String queueType) { + NewTopic newTopic = new NewTopic(topic, numPartitions, replicationFactor); + try { + adminClient.createTopics(Collections.singleton(newTopic)).all().get(); + _log.info("Created topic: {}", topic); + } catch (ExecutionException e) { + if (e.getCause() instanceof TopicExistsException) { + _log.warn("Topic {} already exists.", topic); + } else { + _log.error("Error creating topic {}: {}", topic, e.getMessage()); + } + } catch (InterruptedException e) { + _log.error("Interrupted while creating topic {}: {}", topic, e.getMessage()); + Thread.currentThread().interrupt(); + } + } + + /** + * Determines if a topic already exists in AWS MSK + * @param topic The name of the topic. + */ + public boolean isTopicExists(String topic) { + try { + return adminClient.listTopics().names().get().contains(topic); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } catch (ExecutionException e) { + throw new RuntimeException(e); + } + } + + /** + * Closes the AdminClient to release resources. + */ + public void close() { + adminClient.close(); + } +} \ No newline at end of file diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaConfig.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaConfig.java new file mode 100644 index 0000000000..71e89e87de --- /dev/null +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaConfig.java @@ -0,0 +1,117 @@ +package com.bazaarvoice.emodb.queue.core.kafka; + +import com.amazonaws.AmazonServiceException; +import com.amazonaws.services.simplesystemsmanagement.AWSSimpleSystemsManagement; +import com.amazonaws.services.simplesystemsmanagement.AWSSimpleSystemsManagementClientBuilder; +import com.amazonaws.services.simplesystemsmanagement.model.AWSSimpleSystemsManagementException; +import com.amazonaws.services.simplesystemsmanagement.model.GetParametersRequest; +import com.amazonaws.services.simplesystemsmanagement.model.GetParametersResult; +import com.amazonaws.services.simplesystemsmanagement.model.Parameter; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; + +public class KafkaConfig { + + private static final Logger logger = LoggerFactory.getLogger(KafkaConfig.class); + + // Static SSM Client and configuration using AWS SDK v1 + private static final AWSSimpleSystemsManagement ssmClient = AWSSimpleSystemsManagementClientBuilder + .standard() + .build(); + + private static final String DEFAULT_BOOTSTRAP_SERVERS = + "b-1.qaemodbpocmsk.q4panq.c10.kafka.us-east-1.amazonaws.com:9092," + + "b-2.qaemodbpocmsk.q4panq.c10.kafka.us-east-1.amazonaws.com:9092"; + + private static String bootstrapServersConfig; + private static String batchSizeConfig; + private static String retriesConfig; + private static String lingerMsConfig; + + static { + try { + // Load configurations from SSM during static initialization + Map parameterValues = getParameterValues( + Arrays.asList( + "/emodb/kafka/batchSize", + "/emodb/kafka/retries", + "/emodb/kafka/lingerMs", + "/emodb/kafka/bootstrapServers" + ) + ); + + // Set configurations with fallback to defaults if not present + batchSizeConfig = parameterValues.getOrDefault("/emodb/kafka/batchSize", "16384"); + retriesConfig = parameterValues.getOrDefault("/emodb/kafka/retries", "3"); + lingerMsConfig = parameterValues.getOrDefault("/emodb/kafka/lingerMs", "1"); + bootstrapServersConfig = parameterValues.getOrDefault("/emodb/kafka/bootstrapServers", DEFAULT_BOOTSTRAP_SERVERS); + + logger.info("Kafka configurations loaded successfully from SSM."); + } catch (AmazonServiceException e) { + logger.error("Failed to load configurations from SSM. Using default values.", e); + } + } + + // Fetch parameters from AWS SSM using AWS SDK v1 + private static Map getParameterValues(List parameterNames) { + try { + GetParametersRequest request = new GetParametersRequest() + .withNames(parameterNames) + .withWithDecryption(true); + + GetParametersResult response = ssmClient.getParameters(request); + + return response.getParameters().stream() + .collect(Collectors.toMap(Parameter::getName, Parameter::getValue)); + } catch (AWSSimpleSystemsManagementException e) { + logger.error("Error fetching parameters from SSM.", e); + throw e; // Rethrow or handle the exception if necessary + } + } + + // Kafka Producer properties + public static Properties getProducerProps() { + Properties producerProps = new Properties(); + + producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServersConfig); + producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); + producerProps.put(ProducerConfig.RETRIES_CONFIG, Integer.parseInt(retriesConfig)); + producerProps.put(ProducerConfig.LINGER_MS_CONFIG, Integer.parseInt(lingerMsConfig)); + producerProps.put(ProducerConfig.BATCH_SIZE_CONFIG, Integer.parseInt(batchSizeConfig)); + producerProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // Default buffer memory setting + logger.info("Kafka Producer properties initialized."); + return producerProps; + } + + // Kafka Admin properties + public static Properties getAdminProps() { + Properties adminProps = new Properties(); + + adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServersConfig); + logger.info("Kafka Admin properties initialized."); + return adminProps; + } + + // Ensure the SSM client is closed when the application shuts down + public static void shutdown() { + if (ssmClient != null) { + try { + ssmClient.shutdown(); + logger.info("SSM client closed successfully."); + } catch (Exception e) { + logger.error("Error while closing SSM client.", e); + } + } + } +} diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaProducerService.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaProducerService.java new file mode 100644 index 0000000000..8e55665c42 --- /dev/null +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/kafka/KafkaProducerService.java @@ -0,0 +1,73 @@ +package com.bazaarvoice.emodb.queue.core.kafka; + +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collection; +import java.util.concurrent.Future; + +public class KafkaProducerService { + private static final Logger _log = LoggerFactory.getLogger(KafkaProducerService.class); + private final KafkaProducer producer; // Changed to String + + public KafkaProducerService() { + this.producer = new KafkaProducer<>(KafkaConfig.getProducerProps()); + _log.info("KafkaProducerService initialized with producer properties: {}", KafkaConfig.getProducerProps()); + } + + /** + * Sends each message from the collection to the specified Kafka topic separately. + * + * @param topic The Kafka topic. + * @param events The collection of messages to be sent. + */ + public void sendMessages(String topic, Collection events, String queueType) { + _log.info("Sending {} messages to topic '{}'", events.size(), topic); + for (T event : events) { + _log.debug("Sending message: {}", event); + sendMessage(topic, event.toString(),queueType); + } + _log.info("Finished sending messages to topic '{}'", topic); + } + + /** + * Sends a single message to the specified Kafka topic. + * + * @param topic The Kafka topic. + * @param message The message to be sent. + */ + public void sendMessage(String topic, String message, String queueType) { + ProducerRecord record = new ProducerRecord<>(topic, message); + _log.debug("Preparing to send message to topic '{}' with value: {}", topic, message); + + try { + Future future = producer.send(record, (metadata, exception) -> { + if (exception != null) { + _log.error("Failed to send message to topic '{}'. Error: {}", topic, exception.getMessage()); + } else { + _log.debug("Message sent to topic '{}' partition {} at offset {}", + metadata.topic(), metadata.partition(), metadata.offset()); + } + }); + // Optionally, you can wait for the send to complete + RecordMetadata metadata = future.get(); // Blocking call + _log.info("Message sent successfully to topic '{}' partition {} at offset {}", + metadata.topic(), metadata.partition(), metadata.offset()); + } catch (Exception e) { + _log.error("Failed to send message to topic '{}'. Exception: {}", topic, e.getMessage()); + } + } + + /** + * Closes the producer to release resources. + */ + public void close() { + _log.info("Closing Kafka producer."); + producer.flush(); + producer.close(); + _log.info("Kafka producer closed."); + } +} \ No newline at end of file diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/ssm/ParameterStoreUtil.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/ssm/ParameterStoreUtil.java new file mode 100644 index 0000000000..bf1e6753f0 --- /dev/null +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/ssm/ParameterStoreUtil.java @@ -0,0 +1,112 @@ +package com.bazaarvoice.emodb.queue.core.ssm; + +import com.amazonaws.services.simplesystemsmanagement.AWSSimpleSystemsManagement; +import com.amazonaws.services.simplesystemsmanagement.AWSSimpleSystemsManagementClientBuilder; +import com.amazonaws.services.simplesystemsmanagement.model.GetParameterRequest; +import com.amazonaws.services.simplesystemsmanagement.model.GetParameterResult; +import com.amazonaws.services.simplesystemsmanagement.model.GetParametersRequest; +import com.amazonaws.services.simplesystemsmanagement.model.GetParametersResult; +import com.amazonaws.services.simplesystemsmanagement.model.ParameterNotFoundException; +import com.amazonaws.services.simplesystemsmanagement.model.AWSSimpleSystemsManagementException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Utility class for interacting with AWS Parameter Store using AWS SDK v1. + */ +public class ParameterStoreUtil { + + private static final Logger logger = LoggerFactory.getLogger(ParameterStoreUtil.class); + private final AWSSimpleSystemsManagement ssmClient; + + /** + * Constructor to initialize the SSM client + */ + public ParameterStoreUtil() { + // Create SSM client with default credentials and region + ssmClient = AWSSimpleSystemsManagementClientBuilder.standard() + .build(); + } + + /** + * Fetches a parameter from AWS Parameter Store. + * + * @param parameterName The name of the parameter to fetch + * @return The value of the parameter + * @throws IllegalArgumentException If the parameterName is null or empty + */ + public String getParameter(String parameterName) { + if (parameterName == null || parameterName.isEmpty()) { + logger.error("Parameter name cannot be null or empty"); + throw new IllegalArgumentException("Parameter name cannot be null or empty"); + } + + try { + logger.info("Fetching parameter from AWS Parameter Store: {}", parameterName); + + GetParameterRequest request = new GetParameterRequest().withName(parameterName); + GetParameterResult result = ssmClient.getParameter(request); + + logger.info("Successfully retrieved parameter: {}", parameterName); + return result.getParameter().getValue(); + + } catch (ParameterNotFoundException e) { + logger.error("Parameter not found: {}", parameterName, e); + throw new RuntimeException("Parameter not found: " + parameterName, e); + + } catch (AWSSimpleSystemsManagementException e) { + logger.error("Error fetching parameter from AWS SSM: {}", e.getMessage(), e); + throw new RuntimeException("Error fetching parameter from AWS SSM: " + parameterName, e); + + } catch (Exception e) { + logger.error("Unexpected error while fetching parameter: {}", parameterName, e); + throw new RuntimeException("Unexpected error fetching parameter: " + parameterName, e); + } + } + + /** + * Fetches multiple parameters from AWS Parameter Store in a batch. + * + * @param parameterNames The list of parameter names to fetch + * @return A map of parameter names to their values + * @throws IllegalArgumentException If the parameterNames list is null or empty + */ + public Map getParameters(List parameterNames) { + if (parameterNames == null || parameterNames.isEmpty()) { + logger.error("Parameter names list cannot be null or empty"); + throw new IllegalArgumentException("Parameter names list cannot be null or empty"); + } + + try { + logger.info("Fetching parameters from AWS Parameter Store: {}", parameterNames); + + GetParametersRequest request = new GetParametersRequest().withNames(parameterNames); + GetParametersResult result = ssmClient.getParameters(request); + + // Map the result to a Map of parameter names and values + Map parameters = new HashMap<>(); + result.getParameters().forEach(param -> parameters.put(param.getName(), param.getValue())); + + // Log any parameters that were not found + if (!result.getInvalidParameters().isEmpty()) { + logger.warn("The following parameters were not found: {}", result.getInvalidParameters()); + } + + logger.info("Successfully retrieved {} parameters", parameters.size()); + return parameters; + + } catch (AWSSimpleSystemsManagementException e) { + logger.error("Error fetching parameters from AWS SSM: {}", e.getMessage(), e); + throw new RuntimeException("Error fetching parameters from AWS SSM: " + parameterNames, e); + + } catch (Exception e) { + logger.error("Unexpected error while fetching parameters: {}", parameterNames, e); + throw new RuntimeException("Unexpected error fetching parameters: " + parameterNames, e); + } + } + +} diff --git a/queue/src/main/java/com/bazaarvoice/emodb/queue/core/stepfn/StepFunctionService.java b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/stepfn/StepFunctionService.java new file mode 100644 index 0000000000..bbe04ad17c --- /dev/null +++ b/queue/src/main/java/com/bazaarvoice/emodb/queue/core/stepfn/StepFunctionService.java @@ -0,0 +1,75 @@ +package com.bazaarvoice.emodb.queue.core.stepfn; + + +import com.amazonaws.services.stepfunctions.AWSStepFunctions; +import com.amazonaws.services.stepfunctions.AWSStepFunctionsClientBuilder; +import com.amazonaws.services.stepfunctions.model.StartExecutionRequest; +import com.amazonaws.services.stepfunctions.model.StartExecutionResult; +import com.amazonaws.services.stepfunctions.model.StateMachineDoesNotExistException; +import com.amazonaws.services.stepfunctions.model.InvalidArnException; +import com.amazonaws.services.stepfunctions.model.InvalidExecutionInputException; +import com.amazonaws.services.stepfunctions.model.AWSStepFunctionsException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Production-level service to interact with AWS Step Functions using AWS SDK v1. + */ +public class StepFunctionService { + + private static final Logger logger = LoggerFactory.getLogger(StepFunctionService.class); + + private final AWSStepFunctions stepFunctionsClient; + + /** + * Constructor to initialize Step Function Client with AWS region and credentials. + */ + public StepFunctionService() { + this.stepFunctionsClient = AWSStepFunctionsClientBuilder.standard() + .withRegion("us-east-1") + .build(); + } + + /** + * Starts the execution of a Step Function with the given state machine ARN and input payload. + * + * @param stateMachineArn ARN of the state machine + * @param inputPayload Input for the state machine execution + * @throws IllegalArgumentException If the stateMachineArn is invalid + */ + public void startExecution(String stateMachineArn, String inputPayload) { + if (stateMachineArn == null || stateMachineArn.isEmpty()) { + logger.error("State Machine ARN cannot be null or empty"); + throw new IllegalArgumentException("State Machine ARN cannot be null or empty"); + } + + if (inputPayload == null) { + logger.warn("Input payload is null; using empty JSON object"); + inputPayload = "{}"; // Default to empty payload if null + } + + try { + StartExecutionRequest startExecutionRequest = new StartExecutionRequest() + .withStateMachineArn(stateMachineArn) + .withInput(inputPayload); + + StartExecutionResult startExecutionResult = stepFunctionsClient.startExecution(startExecutionRequest); + + logger.info("Successfully started execution for state machine ARN: {}", stateMachineArn); + logger.debug("Execution ARN: {}", startExecutionResult.getExecutionArn()); + + } catch (StateMachineDoesNotExistException e) { + logger.error("State Machine does not exist: {}", stateMachineArn, e); + } catch (InvalidArnException e) { + logger.error("Invalid ARN provided: {}", stateMachineArn, e); + } catch (InvalidExecutionInputException e) { + logger.error("Invalid execution input provided: {}", inputPayload, e); + } catch (AWSStepFunctionsException e) { + logger.error("Error executing Step Function: {}", e.getMessage(), e); + throw e; // Re-throw after logging + } catch (Exception e) { + logger.error("Unexpected error occurred during Step Function execution: {}", e.getMessage(), e); + throw e; // Re-throw unexpected exceptions + } + } +} diff --git a/sdk/pom.xml b/sdk/pom.xml index e8f8b7eb5b..7ac7c784f7 100644 --- a/sdk/pom.xml +++ b/sdk/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/sor-api/pom.xml b/sor-api/pom.xml index d7bc799c60..61c63331de 100644 --- a/sor-api/pom.xml +++ b/sor-api/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/sor-api/src/main/java/com/bazaarvoice/emodb/sor/api/DataStore.java b/sor-api/src/main/java/com/bazaarvoice/emodb/sor/api/DataStore.java index 7a00eedf9f..e37ef19dac 100644 --- a/sor-api/src/main/java/com/bazaarvoice/emodb/sor/api/DataStore.java +++ b/sor-api/src/main/java/com/bazaarvoice/emodb/sor/api/DataStore.java @@ -261,4 +261,10 @@ void dropFacade(String table, String placement, Audit audit) */ URI getStashRoot() throws StashNotAvailableException; + + default void updateRefInDatabus(Iterable updates, Set tags, boolean isFacade) { + /* + * This method is a no-op in the default implementation. It is used by the Databus to update the reference + */ + } } diff --git a/sor-client-common/pom.xml b/sor-client-common/pom.xml index 8028d79623..0c6673cd47 100644 --- a/sor-client-common/pom.xml +++ b/sor-client-common/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/sor-client-jersey2/pom.xml b/sor-client-jersey2/pom.xml index 4141cd9d99..914276cf4b 100644 --- a/sor-client-jersey2/pom.xml +++ b/sor-client-jersey2/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/sor-client/pom.xml b/sor-client/pom.xml index 100d90210e..91a87d08b9 100644 --- a/sor-client/pom.xml +++ b/sor-client/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/sor/pom.xml b/sor/pom.xml index d2a61758ef..d8158a1576 100644 --- a/sor/pom.xml +++ b/sor/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml @@ -348,5 +348,11 @@ testng test + + com.bazaarvoice.emodb + emodb-queue + 6.5.183-SNAPSHOT + compile + diff --git a/sor/src/main/java/com/bazaarvoice/emodb/sor/core/DefaultDataStore.java b/sor/src/main/java/com/bazaarvoice/emodb/sor/core/DefaultDataStore.java index dadbdd3b4e..46b769fe31 100644 --- a/sor/src/main/java/com/bazaarvoice/emodb/sor/core/DefaultDataStore.java +++ b/sor/src/main/java/com/bazaarvoice/emodb/sor/core/DefaultDataStore.java @@ -5,29 +5,8 @@ import com.bazaarvoice.emodb.common.json.deferred.LazyJsonMap; import com.bazaarvoice.emodb.common.uuid.TimeUUIDs; import com.bazaarvoice.emodb.common.zookeeper.store.MapStore; -import com.bazaarvoice.emodb.sor.api.Audit; -import com.bazaarvoice.emodb.sor.api.AuditBuilder; -import com.bazaarvoice.emodb.sor.api.AuditsUnavailableException; -import com.bazaarvoice.emodb.sor.api.Change; -import com.bazaarvoice.emodb.sor.api.CompactionControlSource; -import com.bazaarvoice.emodb.sor.api.Coordinate; -import com.bazaarvoice.emodb.sor.api.DataStore; -import com.bazaarvoice.emodb.sor.api.DefaultTable; -import com.bazaarvoice.emodb.sor.api.FacadeOptions; -import com.bazaarvoice.emodb.sor.api.History; -import com.bazaarvoice.emodb.sor.api.Intrinsic; -import com.bazaarvoice.emodb.sor.api.Names; -import com.bazaarvoice.emodb.sor.api.ReadConsistency; -import com.bazaarvoice.emodb.sor.api.StashNotAvailableException; -import com.bazaarvoice.emodb.sor.api.StashRunTimeInfo; -import com.bazaarvoice.emodb.sor.api.StashTimeKey; -import com.bazaarvoice.emodb.sor.api.TableOptions; -import com.bazaarvoice.emodb.sor.api.UnknownPlacementException; -import com.bazaarvoice.emodb.sor.api.UnknownTableException; -import com.bazaarvoice.emodb.sor.api.UnpublishedDatabusEvent; -import com.bazaarvoice.emodb.sor.api.UnpublishedDatabusEventType; -import com.bazaarvoice.emodb.sor.api.Update; -import com.bazaarvoice.emodb.sor.api.WriteConsistency; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; +import com.bazaarvoice.emodb.sor.api.*; import com.bazaarvoice.emodb.sor.audit.AuditWriter; import com.bazaarvoice.emodb.sor.compactioncontrol.LocalCompactionControl; import com.bazaarvoice.emodb.sor.condition.Condition; @@ -104,6 +83,7 @@ public class DefaultDataStore implements DataStore, DataProvider, DataTools, Tab private static final int NUM_COMPACTION_THREADS = 2; private static final int MAX_COMPACTION_QUEUE_LENGTH = 100; + public static final String UPDATE_AUDIT_TOPIC = "master_bus"; private final Logger _log = LoggerFactory.getLogger(DefaultDataStore.class); @@ -126,6 +106,7 @@ public class DefaultDataStore implements DataStore, DataProvider, DataTools, Tab private final CompactionControlSource _compactionControlSource; private final MapStore _minSplitSizeMap; private final Clock _clock; + private final KafkaProducerService _kafkaProducerService; private StashTableDAO _stashTableDao; @@ -134,10 +115,10 @@ public DefaultDataStore(LifeCycleRegistry lifeCycle, MetricRegistry metricRegist DataReaderDAO dataReaderDao, DataWriterDAO dataWriterDao, SlowQueryLog slowQueryLog, HistoryStore historyStore, @StashRoot Optional stashRootDirectory, @LocalCompactionControl CompactionControlSource compactionControlSource, @StashBlackListTableCondition Condition stashBlackListTableCondition, AuditWriter auditWriter, - @MinSplitSizeMap MapStore minSplitSizeMap, Clock clock) { + @MinSplitSizeMap MapStore minSplitSizeMap, Clock clock, KafkaProducerService kafkaProducerService) { this(eventWriterRegistry, tableDao, dataReaderDao, dataWriterDao, slowQueryLog, defaultCompactionExecutor(lifeCycle), historyStore, stashRootDirectory, compactionControlSource, stashBlackListTableCondition, auditWriter, - minSplitSizeMap, metricRegistry, clock); + minSplitSizeMap, metricRegistry, clock, kafkaProducerService); } @VisibleForTesting @@ -146,7 +127,7 @@ public DefaultDataStore(DatabusEventWriterRegistry eventWriterRegistry,TableDAO SlowQueryLog slowQueryLog, ExecutorService compactionExecutor, HistoryStore historyStore, Optional stashRootDirectory, CompactionControlSource compactionControlSource, Condition stashBlackListTableCondition, AuditWriter auditWriter, - MapStore minSplitSizeMap, MetricRegistry metricRegistry, Clock clock) { + MapStore minSplitSizeMap, MetricRegistry metricRegistry, Clock clock, KafkaProducerService kafkaProducerService) { _eventWriterRegistry = requireNonNull(eventWriterRegistry, "eventWriterRegistry"); _tableDao = requireNonNull(tableDao, "tableDao"); _dataReaderDao = requireNonNull(dataReaderDao, "dataReaderDao"); @@ -166,6 +147,8 @@ public DefaultDataStore(DatabusEventWriterRegistry eventWriterRegistry,TableDAO _compactionControlSource = requireNonNull(compactionControlSource, "compactionControlSource"); _minSplitSizeMap = requireNonNull(minSplitSizeMap, "minSplitSizeMap"); _clock = requireNonNull(clock, "clock"); + _kafkaProducerService = requireNonNull(kafkaProducerService, "kafkaProducerService"); + } /** @@ -679,17 +662,8 @@ public void updateAll(Iterable updates, Set tags) { } - private void updateAll(Iterable updates, final boolean isFacade, - @NotNull final Set tags) { - requireNonNull(updates, "updates"); - checkLegalTags(tags); - requireNonNull(tags, "tags"); - Iterator updatesIter = updates.iterator(); - if (!updatesIter.hasNext()) { - return; - } - - _dataWriterDao.updateAll(Iterators.transform(updatesIter, new Function() { + private Iterator transformUpdates(Iterator updatesIter, boolean isFacade, final Set tags) { + return Iterators.transform(updatesIter, new Function() { @Override public RecordUpdate apply(Update update) { requireNonNull(update, "update"); @@ -722,7 +696,20 @@ public RecordUpdate apply(Update update) { return new RecordUpdate(table, key, changeId, delta, audit, tags, update.getConsistency()); } - }), new DataWriterDAO.UpdateListener() { + }); + } + + private void updateAll(Iterable updates, final boolean isFacade, + @NotNull final Set tags) { + requireNonNull(updates, "updates"); + checkLegalTags(tags); + requireNonNull(tags, "tags"); + Iterator updatesIter = updates.iterator(); + if (!updatesIter.hasNext()) { + return; + } + + _dataWriterDao.updateAll(transformUpdates(updatesIter, isFacade, tags), new DataWriterDAO.UpdateListener() { @Override public void beforeWrite(Collection updateBatch) { // Tell the databus we're about to write. @@ -744,7 +731,7 @@ public void beforeWrite(Collection updateBatch) { } } if (!updateRefs.isEmpty()) { - _eventWriterRegistry.getDatabusWriter().writeEvents(updateRefs); + _kafkaProducerService.sendMessages(UPDATE_AUDIT_TOPIC, updateRefs, "update"); } } @@ -1025,4 +1012,24 @@ private void decrementDeltaSizes(PendingCompaction pendingCompaction) { private String getMetricName(String name) { return MetricRegistry.name("bv.emodb.sor", "DefaultDataStore", name); } + + @Override + public void updateRefInDatabus(Iterable updates, Set tags, boolean isFacade) { + Iterator updatesIter = updates.iterator(); + if (!updatesIter.hasNext()) { + return; + } + Iterator recordUpdates = transformUpdates(updatesIter, isFacade, tags); + + while (recordUpdates.hasNext()) { + RecordUpdate update = recordUpdates.next(); + List updateRefs = Lists.newArrayListWithCapacity(Collections.singleton(update).size()); + if (!update.getTable().isInternal()) { + updateRefs.add(new UpdateRef(update.getTable().getName(), update.getKey(), update.getChangeId(), tags)); + } + if (!updateRefs.isEmpty()) { + _eventWriterRegistry.getDatabusWriter().writeEvents(updateRefs); + } + } + } } diff --git a/sor/src/main/java/com/bazaarvoice/emodb/sor/core/test/InMemoryDataStore.java b/sor/src/main/java/com/bazaarvoice/emodb/sor/core/test/InMemoryDataStore.java index 37348e976b..a1dd5c09f3 100644 --- a/sor/src/main/java/com/bazaarvoice/emodb/sor/core/test/InMemoryDataStore.java +++ b/sor/src/main/java/com/bazaarvoice/emodb/sor/core/test/InMemoryDataStore.java @@ -1,5 +1,6 @@ package com.bazaarvoice.emodb.sor.core.test; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.audit.DiscardingAuditWriter; import com.bazaarvoice.emodb.sor.compactioncontrol.InMemoryCompactionControlSource; import com.bazaarvoice.emodb.sor.condition.Conditions; @@ -19,18 +20,18 @@ */ public class InMemoryDataStore extends DefaultDataStore { - public InMemoryDataStore(MetricRegistry metricRegistry) { - this(new InMemoryDataReaderDAO(), metricRegistry); + public InMemoryDataStore(MetricRegistry metricRegistry, KafkaProducerService kafkaProducerService) { + this(new InMemoryDataReaderDAO(), metricRegistry, kafkaProducerService); } - public InMemoryDataStore(InMemoryDataReaderDAO dataDao, MetricRegistry metricRegistry) { - this(new DatabusEventWriterRegistry(), dataDao, metricRegistry); + public InMemoryDataStore(InMemoryDataReaderDAO dataDao, MetricRegistry metricRegistry, KafkaProducerService kafkaProducerService) { + this(new DatabusEventWriterRegistry(), dataDao, metricRegistry, kafkaProducerService); } - public InMemoryDataStore(DatabusEventWriterRegistry eventWriterRegistry, InMemoryDataReaderDAO dataDao, MetricRegistry metricRegistry) { + public InMemoryDataStore(DatabusEventWriterRegistry eventWriterRegistry, InMemoryDataReaderDAO dataDao, MetricRegistry metricRegistry, KafkaProducerService kafkaProducerService) { super(eventWriterRegistry, new InMemoryTableDAO(), dataDao, dataDao, new NullSlowQueryLog(), MoreExecutors.newDirectExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), metricRegistry, Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), metricRegistry, Clock.systemUTC(), kafkaProducerService); } } diff --git a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/CompactorTest.java b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/CompactorTest.java index 48da779c69..109f949b48 100644 --- a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/CompactorTest.java +++ b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/CompactorTest.java @@ -1,5 +1,6 @@ package com.bazaarvoice.emodb.sor.core; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.Change; import com.bazaarvoice.emodb.sor.api.ChangeBuilder; @@ -485,7 +486,7 @@ public void compact(Table table, String key, UUID compactionKey, Compaction comp } }; - final DataStore dataStore = new InMemoryDataStore(dataDAO, new MetricRegistry()); + final DataStore dataStore = new InMemoryDataStore(dataDAO, new MetricRegistry(), new KafkaProducerService()); // Create a table for our test dataStore.createTable(tableName, @@ -571,7 +572,7 @@ public Record read(Key key, ReadConsistency ignored) { // Configure the data DAO to read 10 columns initially, causing other column reads to be read lazily dataDAO.setColumnBatchSize(10); - final DataStore dataStore = new InMemoryDataStore(dataDAO, new MetricRegistry()); + final DataStore dataStore = new InMemoryDataStore(dataDAO, new MetricRegistry(), new KafkaProducerService()); // Create a table for our test dataStore.createTable(tableName, diff --git a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/DataStoreTest.java b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/DataStoreTest.java index 3cf9b7b50f..93dd5222af 100644 --- a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/DataStoreTest.java +++ b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/DataStoreTest.java @@ -1,5 +1,6 @@ package com.bazaarvoice.emodb.sor.core; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.Audit; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.Change; @@ -47,7 +48,7 @@ public class DataStoreTest { @Test public void testDeltas() throws Exception { - DataStore store = new InMemoryDataStore(new MetricRegistry()); + DataStore store = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); assertFalse(store.getTableExists(TABLE)); @@ -167,7 +168,7 @@ public void testDeltas() throws Exception { @Test public void testRecordTimestamps() throws Exception { - DataStore store = new InMemoryDataStore(new MetricRegistry()); + DataStore store = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); assertFalse(store.getTableExists(TABLE)); @@ -262,7 +263,7 @@ record = store.get(TABLE, KEY1); @Test public void testRecordTimestampsWithEventTags() throws Exception { - DataStore store = new InMemoryDataStore(new MetricRegistry()); + DataStore store = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); assertFalse(store.getTableExists(TABLE)); diff --git a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/MinSplitSizeTest.java b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/MinSplitSizeTest.java index ac585fa220..6b894ba8a2 100644 --- a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/MinSplitSizeTest.java +++ b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/MinSplitSizeTest.java @@ -1,5 +1,6 @@ package com.bazaarvoice.emodb.sor.core; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.api.TableOptionsBuilder; @@ -43,7 +44,7 @@ public List getSplits(Table table, int recordsPerSplit, int localResplit } }; - DataStore dataStore = new InMemoryDataStore(dataDao, new MetricRegistry()); + DataStore dataStore = new InMemoryDataStore(dataDao, new MetricRegistry(), new KafkaProducerService()); dataStore.createTable("table", new TableOptionsBuilder().setPlacement("default").build(), Collections.emptyMap(), new AuditBuilder().build()); diff --git a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/RedundantDeltaTest.java b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/RedundantDeltaTest.java index 7377838dc5..c073e13e5d 100644 --- a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/RedundantDeltaTest.java +++ b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/RedundantDeltaTest.java @@ -2,6 +2,7 @@ import com.bazaarvoice.emodb.common.json.JsonHelper; import com.bazaarvoice.emodb.common.uuid.TimeUUIDs; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.Audit; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.Change; @@ -65,7 +66,7 @@ public void testRedundantDeltas() throws Exception { DefaultDataStore store = new DefaultDataStore(new DatabusEventWriterRegistry(), new InMemoryTableDAO(), dataDao, dataDao, new NullSlowQueryLog(), new DiscardingExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); store.createTable(TABLE, options, Collections.emptyMap(), newAudit("create table")); @@ -122,7 +123,7 @@ public void testMinUUIDDelta() throws Exception { DefaultDataStore store = new DefaultDataStore(new DatabusEventWriterRegistry(), new InMemoryTableDAO(), dataDao, dataDao, new NullSlowQueryLog(), new DiscardingExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); store.createTable(TABLE, options, Collections.emptyMap(), newAudit("create table")); @@ -159,7 +160,7 @@ public void testRedundancyWithTags() throws Exception { DefaultDataStore store = new DefaultDataStore(new DatabusEventWriterRegistry(), new InMemoryTableDAO(), dataDao, dataDao, new NullSlowQueryLog(), new DiscardingExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); store.createTable(TABLE, options, Collections.emptyMap(), newAudit("create table")); @@ -240,7 +241,7 @@ public void testTagsForNestedMapDeltas() { DefaultDataStore store = new DefaultDataStore(new DatabusEventWriterRegistry(), new InMemoryTableDAO(), dataDao, dataDao, new NullSlowQueryLog(), new DiscardingExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); store.createTable(TABLE, options, Collections.emptyMap(), newAudit("create table")); @@ -260,7 +261,7 @@ public void testRedundancyWithCompactionAndUnchangedTag() throws Exception { DefaultDataStore store = new DefaultDataStore(new DatabusEventWriterRegistry(), new InMemoryTableDAO(), dataDao, dataDao, new NullSlowQueryLog(), new DiscardingExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); store.createTable(TABLE, options, Collections.emptyMap(), newAudit("create table")); @@ -337,7 +338,7 @@ public void testPartialCompactionWithNoRedundancy() throws Exception { DefaultDataStore store = new DefaultDataStore(new DatabusEventWriterRegistry(), tableDao, dataDao, dataDao, new NullSlowQueryLog(), new DiscardingExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); store.createTable(TABLE, options, Collections.emptyMap(), newAudit("create table")); @@ -409,7 +410,7 @@ public void testPartialCompactionWithRedundancy() throws Exception { DefaultDataStore store = new DefaultDataStore(new DatabusEventWriterRegistry(), tableDao, dataDao, dataDao, new NullSlowQueryLog(), new DiscardingExecutorService(), new InMemoryHistoryStore(), Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), new MetricRegistry(), Clock.systemUTC(), new KafkaProducerService()); TableOptions options = new TableOptionsBuilder().setPlacement("default").build(); store.createTable(TABLE, options, Collections.emptyMap(), newAudit("create table")); diff --git a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/SorUpdateTest.java b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/SorUpdateTest.java index 60574f680c..cbb3ed21a7 100644 --- a/sor/src/test/java/com/bazaarvoice/emodb/sor/core/SorUpdateTest.java +++ b/sor/src/test/java/com/bazaarvoice/emodb/sor/core/SorUpdateTest.java @@ -1,6 +1,7 @@ package com.bazaarvoice.emodb.sor.core; import com.bazaarvoice.emodb.common.uuid.TimeUUIDs; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.api.TableOptionsBuilder; @@ -34,7 +35,7 @@ public class SorUpdateTest { public void SetupTest() { final InMemoryDataReaderDAO dataDAO = new InMemoryDataReaderDAO(); _eventWriterRegistry = new DatabusEventWriterRegistry(); - _dataStore = new InMemoryDataStore(_eventWriterRegistry, dataDAO, new MetricRegistry()); + _dataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); // Create a table for our test diff --git a/sor/src/test/java/com/bazaarvoice/emodb/sor/test/MultiDCDataStores.java b/sor/src/test/java/com/bazaarvoice/emodb/sor/test/MultiDCDataStores.java index c67f985342..20def380f2 100644 --- a/sor/src/test/java/com/bazaarvoice/emodb/sor/test/MultiDCDataStores.java +++ b/sor/src/test/java/com/bazaarvoice/emodb/sor/test/MultiDCDataStores.java @@ -1,6 +1,7 @@ package com.bazaarvoice.emodb.sor.test; import com.bazaarvoice.emodb.common.dropwizard.lifecycle.SimpleLifeCycleRegistry; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.audit.DiscardingAuditWriter; import com.bazaarvoice.emodb.sor.compactioncontrol.InMemoryCompactionControlSource; @@ -63,12 +64,12 @@ public MultiDCDataStores(int numDCs, boolean asyncCompacter, MetricRegistry metr if (asyncCompacter) { _stores[i] = new DefaultDataStore(new SimpleLifeCycleRegistry(), metricRegistry, new DatabusEventWriterRegistry(), _tableDao, _inMemoryDaos[i].setHistoryStore(_historyStores[i]), _replDaos[i], new NullSlowQueryLog(), _historyStores[i], - Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), new DiscardingAuditWriter(), new InMemoryMapStore<>(), Clock.systemUTC()); + Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), new DiscardingAuditWriter(), new InMemoryMapStore<>(), Clock.systemUTC(), new KafkaProducerService()); } else { _stores[i] = new DefaultDataStore(new DatabusEventWriterRegistry(), _tableDao, _inMemoryDaos[i].setHistoryStore(_historyStores[i]), _replDaos[i], new NullSlowQueryLog(), MoreExecutors.newDirectExecutorService(), _historyStores[i], Optional.empty(), new InMemoryCompactionControlSource(), Conditions.alwaysFalse(), - new DiscardingAuditWriter(), new InMemoryMapStore<>(), metricRegistry, Clock.systemUTC()); + new DiscardingAuditWriter(), new InMemoryMapStore<>(), metricRegistry, Clock.systemUTC(), new KafkaProducerService()); } } } diff --git a/sor/src/test/java/com/bazaarvoice/emodb/table/db/astyanax/TableLifeCycleTest.java b/sor/src/test/java/com/bazaarvoice/emodb/table/db/astyanax/TableLifeCycleTest.java index 5ad5ff357f..98c5531752 100644 --- a/sor/src/test/java/com/bazaarvoice/emodb/table/db/astyanax/TableLifeCycleTest.java +++ b/sor/src/test/java/com/bazaarvoice/emodb/table/db/astyanax/TableLifeCycleTest.java @@ -11,6 +11,7 @@ import com.bazaarvoice.emodb.common.zookeeper.store.ValueStore; import com.bazaarvoice.emodb.datacenter.api.DataCenter; import com.bazaarvoice.emodb.datacenter.core.DefaultDataCenter; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.Audit; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.FacadeExistsException; @@ -1981,7 +1982,7 @@ dataCenter, mock(RateLimiterCache.class), dataCopyDAO, dataPurgeDAO, } private InMemoryDataStore newBackingStore(MetricRegistry metricRegistry) { - InMemoryDataStore store = new InMemoryDataStore(metricRegistry); + InMemoryDataStore store = new InMemoryDataStore(metricRegistry, new KafkaProducerService()); store.createTable("__system:table", newOptions(PL_GLOBAL), ImmutableMap.of(), newAudit()); store.createTable("__system:table_uuid", newOptions(PL_GLOBAL), ImmutableMap.of(), newAudit()); store.createTable("__system:table_unpublished_databus_events", newOptions(PL_GLOBAL), ImmutableMap.of(), newAudit()); diff --git a/table/pom.xml b/table/pom.xml index e4592dd6b0..09b236c118 100644 --- a/table/pom.xml +++ b/table/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/uac-api/pom.xml b/uac-api/pom.xml index b9ba2a6008..a4e33777f8 100644 --- a/uac-api/pom.xml +++ b/uac-api/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/uac-client-jersey2/pom.xml b/uac-client-jersey2/pom.xml index 38984d2abd..0cdda689a3 100644 --- a/uac-client-jersey2/pom.xml +++ b/uac-client-jersey2/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/uac-client/pom.xml b/uac-client/pom.xml index 55aae7fdb8..47c1120805 100644 --- a/uac-client/pom.xml +++ b/uac-client/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/web-local/pom.xml b/web-local/pom.xml index 4535e14494..f73bcd5d98 100644 --- a/web-local/pom.xml +++ b/web-local/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/web-local/start.sh b/web-local/start.sh index 81d885cb55..a2ce290f58 100755 --- a/web-local/start.sh +++ b/web-local/start.sh @@ -53,7 +53,7 @@ if [[ $# -gt 0 ]]; then ;; --ddl-file) DDL_FILE="${2}" - shift 2 + shif ;; --config-file) CONFIG_FILE="${2}" @@ -71,4 +71,4 @@ if [[ $# -gt 0 ]]; then fi -mvn verify -P init-cassandra,start-emodb -Dconfig.file="${CONFIG_FILE}" -Dddl.file="${DDL_FILE}" -Dpermissions.file="${PERMISSIONS_FILE}" \ No newline at end of file +mvn verify -P init-cassandra,start-emodb -Dconfig.file="${CONFIG_FILE}" -Dddl.file="${DDL_FILE}" -Dpermissions.file="${PERMISSIONS_FILE}" -DskipTests -DskipITs \ No newline at end of file diff --git a/web/pom.xml b/web/pom.xml index eadb8ccbf7..90b3514b4e 100644 --- a/web/pom.xml +++ b/web/pom.xml @@ -6,7 +6,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml diff --git a/web/src/main/java/com/bazaarvoice/emodb/web/resources/blob/BlobStoreResource1.java b/web/src/main/java/com/bazaarvoice/emodb/web/resources/blob/BlobStoreResource1.java index 00c380af6a..e3831e0504 100644 --- a/web/src/main/java/com/bazaarvoice/emodb/web/resources/blob/BlobStoreResource1.java +++ b/web/src/main/java/com/bazaarvoice/emodb/web/resources/blob/BlobStoreResource1.java @@ -54,6 +54,7 @@ import java.util.function.Supplier; import java.util.regex.Pattern; import java.util.stream.StreamSupport; + import static java.lang.String.format; @Path("/blob/1") @@ -96,6 +97,7 @@ public BlobStoreResource1(BlobStore blobStore, Set approvedContentTypes, _blobStore = blobStore; _approvedContentTypes = approvedContentTypes; _metricRegistry = metricRegistry; + _listTableRequestsByApiKey = createMetricCache("listTablesByApiKey"); _createTableRequestsByApiKey = createMetricCache("createTableByApiKey"); _dropTableRequestsByApiKey = createMetricCache("dropTableByApiKey"); @@ -172,6 +174,7 @@ public SuccessResponse createTable(@PathParam("table") String table, if (!subject.hasPermission(Permissions.createBlobTable(resource))) { throw new UnauthorizedException(); } + _blobStore.createTable(table, options, attributes, audit); try { _messagingService.sendCreateTableSQS(table,options,attributes,audit); @@ -382,7 +385,7 @@ public Collection getTablePlacements(@Authenticated Subject subject) { return _blobStore.getTablePlacements(); } - + //change /** * Retrieves the current version of a piece of content from the data store. */ diff --git a/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/DedupQueueResource1.java b/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/DedupQueueResource1.java index c6dcd408fc..34c32181e9 100644 --- a/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/DedupQueueResource1.java +++ b/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/DedupQueueResource1.java @@ -110,6 +110,20 @@ public SuccessResponse sendBatch(@PathParam("queue") String queue, Collection messages) { + // Not partitioned--any server can write messages to Cassandra. + _queueService.sendAll(queue, messages,true); + return SuccessResponse.instance(); + } @POST @Path("_sendbatch") diff --git a/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/QueueResource1.java b/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/QueueResource1.java index ff6334db05..6f5a7b185c 100644 --- a/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/QueueResource1.java +++ b/web/src/main/java/com/bazaarvoice/emodb/web/resources/queue/QueueResource1.java @@ -35,6 +35,7 @@ import javax.ws.rs.WebApplicationException; import javax.ws.rs.core.MediaType; import javax.ws.rs.core.Response; +import java.nio.ByteBuffer; import java.util.Collection; import java.util.List; import java.util.Map; @@ -120,6 +121,22 @@ public SuccessResponse sendBatch(@PathParam("queue") String queue, Collection events) { + //TODO change query param name / type + // Not partitioned--any server can write messages to Cassandra. + _queueService.sendAll(queue, events, true); + return SuccessResponse.instance(); + } + @POST @Path("_sendbatch") @Consumes(MediaType.APPLICATION_JSON) diff --git a/web/src/main/java/com/bazaarvoice/emodb/web/resources/sor/DataStoreResource1.java b/web/src/main/java/com/bazaarvoice/emodb/web/resources/sor/DataStoreResource1.java index 27baf89a04..688e479612 100644 --- a/web/src/main/java/com/bazaarvoice/emodb/web/resources/sor/DataStoreResource1.java +++ b/web/src/main/java/com/bazaarvoice/emodb/web/resources/sor/DataStoreResource1.java @@ -8,19 +8,7 @@ import com.bazaarvoice.emodb.common.json.OrderedJson; import com.bazaarvoice.emodb.common.uuid.TimeUUIDs; import com.bazaarvoice.emodb.datacenter.api.DataCenter; -import com.bazaarvoice.emodb.sor.api.Audit; -import com.bazaarvoice.emodb.sor.api.Change; -import com.bazaarvoice.emodb.sor.api.CompactionControlSource; -import com.bazaarvoice.emodb.sor.api.Coordinate; -import com.bazaarvoice.emodb.sor.api.DataStore; -import com.bazaarvoice.emodb.sor.api.FacadeOptions; -import com.bazaarvoice.emodb.sor.api.Intrinsic; -import com.bazaarvoice.emodb.sor.api.PurgeStatus; -import com.bazaarvoice.emodb.sor.api.Table; -import com.bazaarvoice.emodb.sor.api.TableOptions; -import com.bazaarvoice.emodb.sor.api.UnpublishedDatabusEvent; -import com.bazaarvoice.emodb.sor.api.Update; -import com.bazaarvoice.emodb.sor.api.WriteConsistency; +import com.bazaarvoice.emodb.sor.api.*; import com.bazaarvoice.emodb.sor.core.DataStoreAsync; import com.bazaarvoice.emodb.sor.delta.Delta; import com.bazaarvoice.emodb.sor.delta.Deltas; @@ -776,6 +764,27 @@ public SuccessResponse updateAllForFacade(InputStream in, @QueryParam ("tag") Li return SuccessResponse.instance(); } + @POST + @Path("_updateRef") + @Consumes(MediaType.APPLICATION_JSON) + @Timed(name = "bv.emodb.sor.DataStoreResource1.updateRef", absolute = true) + @ApiOperation(value = "Updates a reference", + notes = "Updates a reference", + response = SuccessResponse.class + ) + public SuccessResponse updateRefToDatabus(InputStream in, + @QueryParam("consistency") @DefaultValue("STRONG") WriteConsistencyParam consistency, + @QueryParam("tag") List tags, + @Authenticated Subject subject) { + + Set tagsSet = (tags == null) ? ImmutableSet.of() : Sets.newHashSet(tags); + Iterable updates = asSubjectSafeUpdateIterable(new JsonStreamingArrayParser<>(in, Update.class), subject, true); + + // Perform the update by writing to Databus + _dataStore.updateRefInDatabus(updates, tagsSet, false); + return SuccessResponse.instance(); + } + /** * Imports an arbitrary size stream of deltas and/or JSON objects. Two formats are supported: array syntax * ('[' object ',' object ',' ... ']') and whitespace-separated objects (object whitespace object whitespace ...) diff --git a/web/src/test/java/com/bazaarvoice/emodb/web/purge/PurgeTest.java b/web/src/test/java/com/bazaarvoice/emodb/web/purge/PurgeTest.java index fd11acab47..2a2804a96e 100644 --- a/web/src/test/java/com/bazaarvoice/emodb/web/purge/PurgeTest.java +++ b/web/src/test/java/com/bazaarvoice/emodb/web/purge/PurgeTest.java @@ -11,6 +11,7 @@ import com.bazaarvoice.emodb.job.handler.DefaultJobHandlerRegistry; import com.bazaarvoice.emodb.job.service.DefaultJobService; import com.bazaarvoice.emodb.queue.api.QueueService; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.Audit; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.CompactionControlSource; @@ -84,7 +85,7 @@ public void setUp() throws Exception { lifeCycleRegistry, _queueService, "testqueue", _jobHandlerRegistry, _jobStatusDAO, _curator, 1, Duration.ZERO, 100, Duration.ofHours(1)); - _store = new InMemoryDataStore(new MetricRegistry()); + _store = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); _dataStoreResource = new DataStoreResource1(_store, new DefaultDataStoreAsync(_store, _service, _jobHandlerRegistry), mock(CompactionControlSource.class), new UnlimitedDataStoreUpdateThrottler()); diff --git a/web/src/test/java/com/bazaarvoice/emodb/web/scanner/ScanUploaderTest.java b/web/src/test/java/com/bazaarvoice/emodb/web/scanner/ScanUploaderTest.java index a96563df8d..44c2359777 100644 --- a/web/src/test/java/com/bazaarvoice/emodb/web/scanner/ScanUploaderTest.java +++ b/web/src/test/java/com/bazaarvoice/emodb/web/scanner/ScanUploaderTest.java @@ -15,6 +15,7 @@ import com.bazaarvoice.emodb.plugin.stash.StashMetadata; import com.bazaarvoice.emodb.plugin.stash.StashStateListener; import com.bazaarvoice.emodb.queue.core.ByteBufferInputStream; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.CompactionControlSource; import com.bazaarvoice.emodb.sor.api.Intrinsic; import com.bazaarvoice.emodb.sor.api.ReadConsistency; @@ -421,7 +422,7 @@ dataTools, scanWriterGenerator, compactionControlSource, mock(LifeCycleRegistry. public void testScanUploadFromExistingScan() throws Exception { MetricRegistry metricRegistry = new MetricRegistry(); // Use an in-memory data store but override the default splits operation to return 4 splits for the test placement - InMemoryDataStore dataStore = spy(new InMemoryDataStore(metricRegistry)); + InMemoryDataStore dataStore = spy(new InMemoryDataStore(metricRegistry, new KafkaProducerService())); when(dataStore.getScanRangeSplits("app_global:default", 1000000, Optional.empty())) .thenReturn(new ScanRangeSplits(ImmutableList.of( createSimpleSplitGroup("00", "40"), @@ -621,7 +622,7 @@ public void testScanFailureRecovery() Lists.newArrayList(), Lists.newArrayList()); InMemoryScanWorkflow scanWorkflow = new InMemoryScanWorkflow(); - ScanStatusDAO scanStatusDAO = new DataStoreScanStatusDAO(new InMemoryDataStore(new MetricRegistry()), "scan_table", "app_global:sys"); + ScanStatusDAO scanStatusDAO = new DataStoreScanStatusDAO(new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()), "scan_table", "app_global:sys"); LocalScanUploadMonitor monitor = new LocalScanUploadMonitor(scanWorkflow, scanStatusDAO, mock(ScanWriterGenerator.class), mock(StashStateListener.class), mock(ScanCountListener.class), mock(DataTools.class), new InMemoryCompactionControlSource(), mock(DataCenters.class)); diff --git a/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreScanStatusDAOTest.java b/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreScanStatusDAOTest.java index 0c19fa2c14..e212ee8267 100644 --- a/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreScanStatusDAOTest.java +++ b/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreScanStatusDAOTest.java @@ -1,6 +1,7 @@ package com.bazaarvoice.emodb.web.scanner.scanstatus; import com.bazaarvoice.emodb.common.uuid.TimeUUIDs; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.AuditBuilder; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.core.test.InMemoryDataStore; @@ -33,7 +34,7 @@ public class DataStoreScanStatusDAOTest { @BeforeMethod public void setUp() { - _dataStore = new InMemoryDataStore(new MetricRegistry()); + _dataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); _dao = new DataStoreScanStatusDAO(_dataStore, "scan_table", "app_global:sys"); } diff --git a/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreStashRequestDAOTest.java b/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreStashRequestDAOTest.java index 4e14f50fd1..4eb62d584d 100644 --- a/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreStashRequestDAOTest.java +++ b/web/src/test/java/com/bazaarvoice/emodb/web/scanner/scanstatus/DataStoreStashRequestDAOTest.java @@ -1,5 +1,6 @@ package com.bazaarvoice.emodb.web.scanner.scanstatus; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.core.test.InMemoryDataStore; import com.codahale.metrics.MetricRegistry; @@ -19,7 +20,7 @@ public class DataStoreStashRequestDAOTest { @BeforeMethod public void setUp() { - _dataStore = new InMemoryDataStore(new MetricRegistry()); + _dataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); _dao = new DataStoreStashRequestDAO(_dataStore, "request_table", "app_global:sys"); } diff --git a/web/src/test/java/com/bazaarvoice/emodb/web/settings/SettingsManagerTest.java b/web/src/test/java/com/bazaarvoice/emodb/web/settings/SettingsManagerTest.java index f8c5758a07..49418a551c 100644 --- a/web/src/test/java/com/bazaarvoice/emodb/web/settings/SettingsManagerTest.java +++ b/web/src/test/java/com/bazaarvoice/emodb/web/settings/SettingsManagerTest.java @@ -3,6 +3,7 @@ import com.bazaarvoice.emodb.cachemgr.api.CacheHandle; import com.bazaarvoice.emodb.cachemgr.api.CacheRegistry; import com.bazaarvoice.emodb.cachemgr.api.InvalidationScope; +import com.bazaarvoice.emodb.queue.core.kafka.KafkaProducerService; import com.bazaarvoice.emodb.sor.api.DataStore; import com.bazaarvoice.emodb.sor.api.Intrinsic; import com.bazaarvoice.emodb.sor.core.test.InMemoryDataStore; @@ -32,7 +33,7 @@ public class SettingsManagerTest { @BeforeMethod public void setUp() { - _dataStore = new InMemoryDataStore(new MetricRegistry()); + _dataStore = new InMemoryDataStore(new MetricRegistry(), new KafkaProducerService()); _cacheRegistry = mock(CacheRegistry.class); _cacheHandle = mock(CacheHandle.class); when(_cacheRegistry.register(eq("settings"), any(Cache.class), eq(true))).thenReturn(_cacheHandle); diff --git a/yum/pom.xml b/yum/pom.xml index adee7cf547..e633838c55 100644 --- a/yum/pom.xml +++ b/yum/pom.xml @@ -4,7 +4,7 @@ com.bazaarvoice.emodb emodb-parent - 6.5.171-SNAPSHOT + 6.5.184-SNAPSHOT ../parent/pom.xml