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.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileCodec.java:
##########
@@ -26,55 +26,56 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ByteBuffInputStream;
import org.apache.hadoop.hbase.nio.ByteBuff;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.WritableUtils;
/**
- * Default Codec for encoding and decoding ReplicationLog Records within a
block buffer. This
- * implementation uses standard Java DataInput/DataOutput for serialization.
Record Format within a
- * block:
+ * Default Codec for encoding and decoding ReplicationLog Records within a
block buffer. The on-disk
Review Comment:
This is the most interesting change (to me). I suppose it's of a piece with
the evolution of HBase WAL with WALEdit to group cells and on the RPC side
"cell blocks" instead of individually described cells in the IDL.
We may want to restructure the ascii art here further
```text
RECORD LENGTH (vint)
RECORD HEADER
Table name length (vint)
Table name bytes
Commit id (vlong)
ATTRIBUTES COUNT (vint) -- NEW: record-level attribute map
PER-ATTRIBUTE (repeated):
Key length (vint)
Key bytes (UTF-8)
Value length (vint)
Value bytes
CELL COUNT (vint)
PER-CELL (repeated):
Row length (vint) -- row written on EVERY cell
Row bytes
Family length (vint) -- family written on EVERY cell
Family bytes
Qualifier length (vint)
Qualifier bytes
Cell timestamp (long, 8 bytes)
Cell type byte (1 byte)
Value length (vint)
Value bytes
```
##########
phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java:
##########
@@ -1429,59 +1456,15 @@ public void testRetryPicksUpStagedWriter() throws
Exception {
assertTrue("Should be using new writer", newWriter != initialWriter);
// Old writer: received the append only
- verify(initialWriter, times(1)).append(eq(tableName), eq(commitId),
eq(put));
+ verify(initialWriter, times(1)).append(eq(tableName), eq(commitId),
+ eq(LogFileTestUtil.cellsOf(put)));
// New writer: received replayed append + successful sync
- verify(newWriter, times(1)).append(eq(tableName), eq(commitId), eq(put));
+ verify(newWriter, times(1)).append(eq(tableName), eq(commitId),
+ eq(LogFileTestUtil.cellsOf(put)));
verify(newWriter, times(1)).sync();
}
- /**
- * Tests the idle-then-lease-recovery scenario: after a sync clears
currentBatch, the system goes
- * idle. A rotation tick stages a pending writer. The reader performs HDFS
lease recovery,
- * breaking the old writer's stream. When events resume, apply() drains the
healthy staged writer
- * before the action — the broken writer is never touched. No replay needed
(empty batch).
- */
- @Test
- public void testIdleLeaseRecoveryDrainsStagedWriter() throws Exception {
Review Comment:
I think this test is a real concurrency concern. After a sync clears
`currentBatch`, the system goes idle, a rotation tick stages a pending writer,
and the reader performs HDFS lease recovery, invalidating the old writer's
outputstream. When events resume we need to ensure `apply()` drains to the new
and healthy writer before the action and should ensure the broken writer is
never touched.
The body of this test only needs some simple updates (any(Mutation.class) →
any(List.class), eq(put) → eq(LogFileTestUtil.cellsOf(put))) Let's restore and
migrate it.
--
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]
