apurtell commented on code in PR #2540:
URL: https://github.com/apache/phoenix/pull/2540#discussion_r3500554618
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java:
##########
@@ -246,21 +246,19 @@ public void processLogFile(FileSystem fs, Path filePath)
throws IOException {
for (LogFile.Record record : logFileReader) {
final TableName tableName =
TableName.valueOf(record.getHBaseTableName());
- final Mutation mutation = record.getMutation();
-
- tableToMutationsMap.computeIfAbsent(tableName, k -> new
ArrayList<>()).add(mutation);
-
- // Increment current batch size and current batch size bytes
- currentBatchSize++;
- currentBatchSizeBytes += mutation.heapSize();
-
- // Process when we reach either the batch count or size limit
- if (currentBatchSize >= getBatchSize() || currentBatchSizeBytes >=
getBatchSizeBytes()) {
- processReplicationLogBatch(tableToMutationsMap);
- totalProcessed += currentBatchSize;
- tableToMutationsMap.clear();
- currentBatchSize = 0;
- currentBatchSizeBytes = 0;
+ for (Mutation mutation : record.getMutations()) {
+ tableToMutationsMap.computeIfAbsent(tableName, k -> new
ArrayList<>()).add(mutation);
+ currentBatchSize++;
+ currentBatchSizeBytes += mutation.heapSize();
+
+ // Process when we reach either the batch count or size limit
Review Comment:
Before this change a record was always a single Mutation, so batches always
split between records. Now a single record's mutations can be split across two
`processReplicationLogBatch` invocations. I think it's fine but future
maintainers cannot assume a certain atomicity here.
--
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]
