Skip to content

Commit 72d0126

Browse files
Ian-NaraIan-Nara
committed
updating comment format
1 parent 0a6f8fa commit 72d0126

File tree

7 files changed

+27
-32
lines changed

7 files changed

+27
-32
lines changed

‎src/main/java/com/uid2/optout/Main.java‎

Lines changed: 10 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -274,38 +274,33 @@ public void run(String[] args) throws IOException {
274274

275275
// deploy sqs producer if enabled
276276
if (this.enqueueSqsEnabled) {
277-
LOGGER.info("SQS enabled, deploying OptOutSqsLogProducer");
277+
LOGGER.info("sqs enabled, deploying OptOutSqsLogProducer");
278278
try {
279-
// create sqs-specific cloud sync with custom folder (default: "sqs-delta")
279+
// sqs delta production uses a separate s3 folder (default: "sqs-delta")
280+
// OptOutCloudSync reads from optout_s3_folder, so we override it with optout_sqs_s3_folder
280281
String sqsFolder = this.config.getString(Const.Config.OptOutSqsS3FolderProp, "sqs-delta");
281-
LOGGER.info("sqs config - optout_sqs_s3_folder: {}, will override optout_s3_folder to: {}",
282-
sqsFolder, sqsFolder);
283-
JsonObject sqsConfig = new JsonObject().mergeIn(this.config)
282+
JsonObject sqsCloudSyncConfig = new JsonObject().mergeIn(this.config)
284283
.put(Const.Config.OptOutS3FolderProp, sqsFolder);
285-
LOGGER.info("sqs config after merge - optout_s3_folder: {}", sqsConfig.getString(Const.Config.OptOutS3FolderProp));
286-
OptOutCloudSync sqsCs = new OptOutCloudSync(sqsConfig, true);
284+
OptOutCloudSync sqsCs = new OptOutCloudSync(sqsCloudSyncConfig, true);
287285

288-
// create sqs-specific cloud storage instance (same bucket, different folder handling)
286+
// create cloud storage instances
289287
ICloudStorage fsSqs;
290288
boolean useStorageMock = this.config.getBoolean(Const.Config.StorageMockProp, false);
291289
if (useStorageMock) {
292-
// reuse the same LocalStorageMock for testing
293290
fsSqs = this.fsOptOut;
294291
} else {
295-
// create fresh CloudStorage for SQS (no path conversion wrapper)
296292
String optoutBucket = this.config.getString(Const.Config.OptOutS3BucketProp);
297-
fsSqs = CloudUtils.createStorage(optoutBucket, sqsConfig);
293+
fsSqs = CloudUtils.createStorage(optoutBucket, this.config);
298294
}
299295

300-
// create sqs-specific cloud storage instance for dropped requests (different bucket)
301296
String optoutBucketDroppedRequests = this.config.getString(Const.Config.OptOutS3BucketDroppedRequestsProp);
302-
ICloudStorage fsSqsDroppedRequests = CloudUtils.createStorage(optoutBucketDroppedRequests, config);
297+
ICloudStorage fsSqsDroppedRequests = CloudUtils.createStorage(optoutBucketDroppedRequests, this.config);
303298

304-
// deploy sqs log producer with its own storage instances
299+
// deploy sqs log producer
305300
OptOutSqsLogProducer sqsLogProducer = new OptOutSqsLogProducer(this.config, fsSqs, fsSqsDroppedRequests, sqsCs, Const.Event.DeltaProduce, null);
306301
futs.add(this.deploySingleInstance(sqsLogProducer));
307302

308-
LOGGER.info("SQS log producer deployed - bucket: {}, folder: {}",
303+
LOGGER.info("sqs log producer deployed, bucket={}, folder={}",
309304
this.config.getString(Const.Config.OptOutS3BucketProp), sqsFolder);
310305
} catch (IOException e) {
311306
LOGGER.error("circuit_breaker_config_error: failed to initialize sqs log producer, delta production will be disabled: {}", e.getMessage(), e);

‎src/main/java/com/uid2/optout/delta/DeltaFileWriter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ private void flushToStream(ByteArrayOutputStream stream) throws IOException {
105105
private void ensureCapacity(int dataSize) {
106106
if (buffer.capacity() < dataSize) {
107107
int newCapacity = Integer.highestOneBit(dataSize) << 1;
108-
LOGGER.info("expanding buffer size: current {}, need {}, new {}", buffer.capacity(), dataSize, newCapacity);
108+
LOGGER.info("expanding buffer: currentSize={}, neededSize={}, newSize={}", buffer.capacity(), dataSize, newCapacity);
109109
this.buffer = ByteBuffer.allocate(newCapacity).order(ByteOrder.LITTLE_ENDIAN);
110110
}
111111
}

‎src/main/java/com/uid2/optout/sqs/SqsBatchProcessor.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -86,10 +86,10 @@ public List<SqsParsedMessage> getMessages() {
8686
* @return BatchProcessingResult containing eligible messages and processing metadata
8787
*/
8888
public BatchProcessingResult processBatch(List<Message> messageBatch, int batchNumber) throws IOException {
89-
// Parse and sort messages by timestamp
89+
// parse and sort messages by timestamp
9090
List<SqsParsedMessage> parsedBatch = SqsMessageParser.parseAndSortMessages(messageBatch);
9191

92-
// Identify and delete corrupt messages
92+
// identify and delete corrupt messages
9393
if (parsedBatch.size() < messageBatch.size()) {
9494
List<Message> invalidMessages = identifyInvalidMessages(messageBatch, parsedBatch);
9595
if (!invalidMessages.isEmpty()) {
@@ -98,21 +98,21 @@ public BatchProcessingResult processBatch(List<Message> messageBatch, int batchN
9898
}
9999
}
100100

101-
// No valid messages after deleting corrupt ones, continue reading
101+
// no valid messages after deleting corrupt ones, continue reading
102102
if (parsedBatch.isEmpty()) {
103103
LOGGER.info("no valid messages in batch {} after removing invalid messages", batchNumber);
104104
return BatchProcessingResult.corruptMessagesDeleted();
105105
}
106106

107-
// Check if the oldest message in this batch is too recent
107+
// check if the oldest message in this batch is too recent
108108
long currentTime = OptOutUtils.nowEpochSeconds();
109109
SqsParsedMessage oldestMessage = parsedBatch.get(0);
110110

111111
if (!isMessageEligible(oldestMessage, currentTime)) {
112112
return BatchProcessingResult.messagesTooRecent();
113113
}
114114

115-
// Filter for eligible messages (>= deltaWindowSeconds old)
115+
// filter for eligible messages (>= deltaWindowSeconds old)
116116
List<SqsParsedMessage> eligibleMessages = filterEligibleMessages(parsedBatch, currentTime);
117117

118118
return BatchProcessingResult.withMessages(eligibleMessages);

‎src/main/java/com/uid2/optout/sqs/SqsMessageOperations.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ public static QueueAttributes getQueueAttributes(SqsClient sqsClient, String que
8787
int delayed = parseIntOrDefault(attrs.get(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES_DELAYED), 0);
8888

8989
QueueAttributes queueAttributes = new QueueAttributes(visible, invisible, delayed);
90-
LOGGER.info("queue attributes: {}", queueAttributes);
90+
LOGGER.info("sqs_queue_attributes={}", queueAttributes);
9191
return queueAttributes;
9292

9393
} catch (Exception e) {

‎src/main/java/com/uid2/optout/sqs/SqsWindowReader.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public static class WindowReadResult {
5050
private final List<SqsParsedMessage> messages;
5151
private final long windowStart;
5252
private final StopReason stopReason;
53-
private final int rawMessagesRead; // total messages pulled from SQS
53+
private final int rawMessagesRead; // total messages pulled from sqs
5454

5555
private WindowReadResult(List<SqsParsedMessage> messages, long windowStart, StopReason stopReason, int rawMessagesRead) {
5656
this.messages = messages;
@@ -97,15 +97,15 @@ public WindowReadResult readWindow() throws IOException {
9797
List<SqsParsedMessage> windowMessages = new ArrayList<>();
9898
long currentWindowStart = 0;
9999
int batchNumber = 0;
100-
int rawMessagesRead = 0; // track total messages pulled from SQS
100+
int rawMessagesRead = 0; // track total messages pulled from sqs
101101

102102
while (true) {
103103
if (windowMessages.size() >= maxMessagesPerWindow) {
104104
LOGGER.warn("high_message_volume: message limit exceeded while reading window, {} messages >= limit {}", windowMessages.size(), maxMessagesPerWindow);
105105
return WindowReadResult.messageLimitExceeded(windowMessages, currentWindowStart, rawMessagesRead);
106106
}
107107

108-
// Read one batch from SQS (up to 10 messages)
108+
// read one batch from sqs (up to 10 messages)
109109
List<Message> rawBatch = SqsMessageOperations.receiveMessagesFromSqs(
110110
this.sqsClient, this.queueUrl, this.maxMessagesPerPoll, this.visibilityTimeout);
111111

@@ -122,21 +122,21 @@ public WindowReadResult readWindow() throws IOException {
122122
if (batchResult.getStopReason() == StopReason.MESSAGES_TOO_RECENT) {
123123
return WindowReadResult.messagesTooRecent(windowMessages, currentWindowStart, rawMessagesRead);
124124
}
125-
// Corrupt messages were deleted, continue reading
125+
// corrupt messages were deleted, continue reading
126126
continue;
127127
}
128128

129-
// Add eligible messages to current window
129+
// add eligible messages to current window
130130
boolean newWindow = false;
131131
for (SqsParsedMessage msg : batchResult.getMessages()) {
132132
long msgWindowStart = msg.timestamp();
133133

134-
// Discover start of window
134+
// discover start of window
135135
if (currentWindowStart == 0) {
136136
currentWindowStart = msgWindowStart;
137137
}
138138

139-
// Discover next window
139+
// discover next window
140140
if (msgWindowStart > currentWindowStart + this.deltaWindowSeconds) {
141141
newWindow = true;
142142
}

‎src/main/java/com/uid2/optout/traffic/TrafficCalculator.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -322,8 +322,8 @@ public TrafficStatus calculateStatus(List<SqsParsedMessage> sqsMessages, QueueAt
322322
}
323323

324324
/**
325-
* find the newest timestamp from delta files.
326-
* reads the newest delta file and returns its maximum timestamp.
325+
* Find the newest timestamp from delta files.
326+
* Reads the newest delta file and returns its maximum timestamp.
327327
*/
328328
private long findNewestDeltaTimestamp(List<String> deltaS3Paths) throws IOException {
329329
if (deltaS3Paths == null || deltaS3Paths.isEmpty()) {

‎src/main/java/com/uid2/optout/vertx/OptOutServiceVerticle.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -402,7 +402,7 @@ private void handleQueue(RoutingContext routingContext) {
402402
.put("email", email)
403403
.put("phone", phone);
404404

405-
// send message to SQS queue
405+
// send message to sqs queue
406406
vertx.executeBlocking(() -> {
407407
// check queue size limit before sending
408408
if (this.sqsMaxQueueSize > 0) {

0 commit comments

Comments
 (0)