rmahindra123 commented on code in PR #13836:
URL: https://github.com/apache/hudi/pull/13836#discussion_r2323604051
##########
hudi-aws/src/main/java/org/apache/hudi/aws/transaction/lock/S3StorageLockClient.java:
##########
@@ -267,6 +269,52 @@ private static S3Client createS3Client(Region region, long
timeoutSecs, Properti
.region(region).build();
}
+ @Override
+ public Option<String> readObject(String filePath, boolean checkExistsFirst) {
+ try {
+ // Parse the file path to get bucket and key
+ URI uri = new URI(filePath);
+ String bucket = uri.getHost();
+ String key = uri.getPath().replaceFirst("/", "");
+
+ if (checkExistsFirst) {
+ // First check if the file exists (lightweight HEAD request)
+ try {
+ s3Client.headObject(HeadObjectRequest.builder()
+ .bucket(bucket)
+ .key(key)
+ .build());
+ } catch (S3Exception e) {
+ if (e.statusCode() == NOT_FOUND_ERROR_CODE) {
+ // File doesn't exist - this is the common case for optional
configs
+ logger.debug("JSON config file not found: {}", filePath);
+ return Option.empty();
+ }
+ throw e; // Re-throw other errors
+ }
+ }
+
+ // Read the file (either after existence check or directly)
+ byte[] bytes = s3Client.getObjectAsBytes(
Review Comment:
can't we do asUtf8String directly?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java:
##########
@@ -590,4 +612,45 @@ private void logErrorLockState(LockState state, String
msg) {
long getCurrentEpochMs() {
return System.currentTimeMillis();
}
+
+ /**
+ * Helper method to start audit operation when lock is acquired.
+ */
+ private void startAuditOperation(long timestamp) {
+ storageLpAuditService.ifPresent(auditService -> {
+ try {
+ String sessionId = UUID.randomUUID().toString();
+ auditService.startOperation(sessionId, timestamp);
+ } catch (Exception e) {
+ // Log but don't fail the lock operation due to recording failures
+ logger.warn("Owner {}: Failed to start audit operation: {}", ownerId,
e.getMessage());
+ }
+ });
+ }
+
+ /**
+ * Helper method to update audit operation for renewals.
+ */
+ private void updateAuditOperation(long timestamp) {
+ storageLpAuditService.ifPresent(auditService -> {
+ try {
+ auditService.updateOperation(timestamp);
+ } catch (Exception e) {
+ logger.warn("Owner {}: Failed to update audit operation: {}", ownerId,
e.getMessage());
+ }
+ });
+ }
+
+ /**
+ * Helper method to end audit operation.
+ */
+ private void endAuditOperation(long timestamp, String finalState) {
Review Comment:
wondering if we should just have enums for well-defined states, and then
only a single API that has the enum state (e.g., RENEW) with the timestamp?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java:
##########
@@ -97,6 +99,7 @@ public class StorageBasedLockProvider implements
LockProvider<StorageLockFile> {
private final HeartbeatManager heartbeatManager;
private final transient Thread shutdownThread;
private final Option<HoodieLockMetrics> hoodieLockMetrics;
+ private final Option<AuditService> storageLpAuditService;
Review Comment:
nit storageLpAuditService -> auditService, i guess its already clear its a
storage lock provider.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java:
##########
@@ -590,4 +612,45 @@ private void logErrorLockState(LockState state, String
msg) {
long getCurrentEpochMs() {
return System.currentTimeMillis();
}
+
+ /**
+ * Helper method to start audit operation when lock is acquired.
+ */
+ private void startAuditOperation(long timestamp) {
+ storageLpAuditService.ifPresent(auditService -> {
Review Comment:
nit: can u write a static method to do this one time?
private void withAuditService(Consumer<AuditService> action) {
auditService.ifPresent(auditService -> {
try {
action.accept(auditService);
} catch (Exception e) {
logger.warn("Owner {}: Failed to update audit operation: {}",
ownerId, e.getMessage());
}
});
}
##########
hudi-aws/src/main/java/org/apache/hudi/aws/transaction/lock/S3StorageLockClient.java:
##########
@@ -267,6 +269,52 @@ private static S3Client createS3Client(Region region, long
timeoutSecs, Properti
.region(region).build();
}
+ @Override
+ public Option<String> readObject(String filePath, boolean checkExistsFirst) {
+ try {
+ // Parse the file path to get bucket and key
+ URI uri = new URI(filePath);
+ String bucket = uri.getHost();
+ String key = uri.getPath().replaceFirst("/", "");
+
+ if (checkExistsFirst) {
Review Comment:
do u think this is required? does nt it double the cost, was wondering if we
just GET the object.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
