diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java index 694bd310612..f2f6901ff70 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java @@ -22,7 +22,6 @@ import java.net.URI; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -78,9 +77,10 @@ public class ReplicationLog { protected final AtomicLong rotationFailures = new AtomicLong(0); // Staged writer created by the background LogRotationTask, drained by checkAndReplaceWriter(). private final AtomicReference pendingWriter = new AtomicReference<>(); - // Latch set by apply() on the retry path before calling requestRotation(); counted down by - // LogRotationTask in a finally block so apply() can wait (with timeout) for a fresh writer. - private volatile CountDownLatch rotationStagedLatch; + // Monitor the apply() retry path waits on for a fresh writer to be staged. LogRotationTask + // notifies it in a finally block on every completion (success or failure) so a waiter is never + // stranded. The waited-on condition is pendingWriter itself, so a spurious notify is harmless. + private final Object rotationSignal = new Object(); private final AtomicBoolean closed = new AtomicBoolean(false); // Single gate for rotation submission. Set by requestRotation()'s CAS before queuing a task, // cleared in LogRotationTask's finally. Both scheduled ticks and on-demand callers go through @@ -251,11 +251,14 @@ protected void checkAndReplaceWriter(boolean asyncClose) { * current writer stays open so in-flight writes still land. Skipping ahead of the CAS (rather * than inside {@link LogRotationTask#run()}) keeps the gate clear, so a later tick resumes * rotation as soon as the flag clears on abort. + * @return {@code true} if a rotation is now queued or already in flight (worth waiting for); + * {@code false} if rotation is suppressed this call (failover pending, or the executor is + * shutting down) so no task will run. */ - private void requestRotation() { + private boolean requestRotation() { if (logGroup.isFailoverPending()) { LOG.info("HAGroup {} rotation suspended: failover pending", logGroup); - return; + return false; } if (rotationRequested.compareAndSet(false, true)) { try { @@ -263,7 +266,45 @@ private void requestRotation() { } catch (java.util.concurrent.RejectedExecutionException e) { LOG.info("Rotation executor shut down, skipping rotation", e); rotationRequested.set(false); + return false; + } + } + return true; + } + + /** + * Requests rotation and waits, bounded by {@code retryDelayMs}, for {@link LogRotationTask} to + * stage a fresh writer in {@code pendingWriter}. Called only from {@link #apply}'s retry path so + * a failed write is retried on a brand-new writer (new HDFS pipeline), never the fenced one. + *

+ * Each spin re-issues {@link #requestRotation()} before waiting: a request coalesced away while a + * soon-to-complete rotation held the CAS gate is reissued once the gate clears, so a fresh task + * actually gets scheduled. The waited-on condition is {@code pendingWriter} itself, so a spurious + * or unrelated notify just re-checks and loops. Exits early if rotation is permanently suppressed + * (nothing will ever stage) or the log closes. + * @return the staged writer, or {@code null} if none was staged before the deadline / close / + * permanent suppression. The caller drains it via {@link #checkAndReplaceWriter}. + */ + private LogFileWriter awaitStagedWriter() throws InterruptedIOException { + final long deadlineNs = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(retryDelayMs); + synchronized (rotationSignal) { + LogFileWriter staged; + while ((staged = pendingWriter.get()) == null && !isClosed()) { + long remainingMs = TimeUnit.NANOSECONDS.toMillis(deadlineNs - System.nanoTime()); + if (remainingMs <= 0) { + break; + } + if (!requestRotation()) { + break; + } + try { + rotationSignal.wait(remainingMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while awaiting a fresh writer"); + } } + return staged; } } @@ -339,20 +380,28 @@ private void apply(Action action) throws IOException { action.action(currentWriter); break; } catch (IOException e) { - LOG.debug("Attempt {}/{} failed", attempt, maxAttempts, e); + // Exhausted: propagate without logging here. The caller (LogEventHandler#onEvent) logs the + // failure with cause and drives the SYNC->SAF transition, so re-logging would duplicate it. if (attempt == maxAttempts) { throw e; } - // Each retry runs on a fresh writer. Stage a latch, request rotation, and wait briefly - // for the LogRotationTask to count the latch down after staging a new pendingWriter. - CountDownLatch latch = new CountDownLatch(1); - rotationStagedLatch = latch; - requestRotation(); - try { - latch.await(retryDelayMs, TimeUnit.MILLISECONDS); - } catch (InterruptedException ie) { - Thread.currentThread().interrupt(); - throw new InterruptedIOException("Interrupted during retry delay"); + // A retry is only useful on a FRESH writer. The current writer is fenced by this failure + // (see LogFileWriter) and, like an HDFS stream that failed a sync, cannot be re-driven -- + // retrying on it would just re-throw. So request rotation and wait briefly for the + // LogRotationTask to stage a new pendingWriter (created off this thread so a slow standby + // FS cannot stall event processing beyond the bounded wait). + // WARN with cause: if the retry below succeeds nothing propagates, so this is the only + // record of the transient failure and must carry the stack. + LOG.warn("Write attempt {}/{} failed on writer {}; requesting rotation to retry on a fresh" + + " writer", attempt, maxAttempts, currentWriter, e); + if (awaitStagedWriter() == null) { + // No fresh writer staged, so there is nothing new to retry on. Surface the original + // failure rather than burning the next attempt on the fenced writer. Message-only: the + // cause was logged with its stack just above, and LogRotationTask logs any + // createNewWriter() failure with its stack separately. + LOG.warn("No fresh writer staged within {}ms; surfacing original failure rather than" + + " retrying fenced writer {}", retryDelayMs, currentWriter); + throw e; } } } @@ -477,10 +526,13 @@ public void run() { logGroup.getMetrics().updateRotationTime(System.nanoTime() - startNs); // Clear last so requestRotation()'s CAS suppresses duplicates throughout this run. rotationRequested.set(false); - CountDownLatch latch = rotationStagedLatch; - if (latch != null) { - latch.countDown(); - rotationStagedLatch = null; + // Wake any apply() retry waiting for a fresh writer. Fires on both success and failure so a + // waiter is never stranded: on failure it wakes, sees pendingWriter still null with the + // gate + // cleared, and either re-drives or times out. Notify after clearing the gate so a woken + // waiter's requestRotation() re-drive is not suppressed by this run's own flag. + synchronized (rotationSignal) { + rotationSignal.notifyAll(); } if (staged) { // Wake an idle consumer so it drains pendingWriter before the reader's round buffer diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java index d4218d69da5..add7d514d83 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java @@ -389,6 +389,19 @@ public void setValues(int type, Record record, CompletableFuture syncFutur this.syncFuture = syncFuture; this.timestampNs = System.nanoTime(); } + + static String typeName(int type) { + switch (type) { + case EVENT_TYPE_DATA: + return "DATA"; + case EVENT_TYPE_SYNC: + return "SYNC"; + case EVENT_TYPE_SWAP: + return "SWAP"; + default: + return "UNKNOWN(" + type + ")"; + } + } } /** @@ -1532,8 +1545,8 @@ public void onEvent(LogEvent event, long sequence, boolean endOfBatch) throws Ex } } catch (IOException e) { try { - LOG.info("Failed to process event at sequence {} on mode {}", sequence, currentModeImpl, - e); + LOG.info("Failed to process {} event at sequence {} on mode {}", + LogEvent.typeName(event.type), sequence, currentModeImpl, e); onFailure(event, sequence, e); } catch (Exception fatalEx) { IOException fatalIOE = diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileWriter.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileWriter.java index 3404fe0baa2..b7122ec2d40 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileWriter.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileWriter.java @@ -40,6 +40,15 @@ public class LogFileWriter implements LogFile.Writer { private LogFileWriterContext context; private LogFileFormatWriter writer; private final AtomicBoolean closed = new AtomicBoolean(false); + /** + * Latched on the first append/sync failure. Once set, the writer is fenced: every subsequent + * append/sync rethrows it rather than touching the underlying stream. This mirrors HDFS + * DFSOutputStream semantics -- once a sync fails the stream tears itself down and any further + * call throws. A partially-written block cannot be safely re-driven on the same writer, so the + * higher layer (ReplicationLog) must recover by rotating to a fresh writer and replaying the + * unsynced batch, not by retrying on this instance. + */ + private volatile IOException fault; /** * A monotonically increasing sequence number that identifies this writer instance, used to detect * log file rotations and ensure proper handling of in-flight operations. Higher layers will get a @@ -89,19 +98,41 @@ public boolean append(String tableName, long commitId, List cells) throws @Override public boolean append(String tableName, long commitId, List cells, Map attributes) throws IOException { - if (isClosed()) { - throw new IOException("Writer has been closed"); + checkWritable(); + try { + return writer.append(new LogFileRecord().setHBaseTableName(tableName).setCommitId(commitId) + .setCells(cells).setAttributes(attributes)); + } catch (IOException e) { + throw fault(e); } - return writer.append(new LogFileRecord().setHBaseTableName(tableName).setCommitId(commitId) - .setCells(cells).setAttributes(attributes)); } @Override public void sync() throws IOException { + checkWritable(); + try { + writer.sync(); + } catch (IOException e) { + throw fault(e); + } + } + + /** Throws if the writer is closed or has been fenced by a prior append/sync failure. */ + private void checkWritable() throws IOException { if (isClosed()) { throw new IOException("Writer has been closed"); } - writer.sync(); + if (fault != null) { + throw new IOException("Writer is faulted by a prior failure", fault); + } + } + + /** Latches the first failure so subsequent append/sync calls fail fast, then returns it. */ + private IOException fault(IOException e) { + if (fault == null) { + fault = e; + } + return e; } @Override diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java index 4ebb8c3c078..855fe9dccb0 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java @@ -801,6 +801,47 @@ public void testSwitchToStoreAndForwardOnSyncFailure() throws Exception { assertEquals(STORE_AND_FORWARD, logGroup.getMode()); } + /** + * S17b: when a sync fails AND rotation cannot stage a fresh writer (the same fault also blocks + * createNewWriter()), apply() must not spend its second attempt re-syncing the fenced writer. It + * surfaces the failure after attempt 1, and the mode still flips to STORE_AND_FORWARD. This is + * the companion to {@link #testSwitchToStoreAndForwardOnSyncFailure()} (where rotation DOES stage + * a writer, so the retry runs on it): here rotation is dead, so there is no second sync. + */ + @Test + public void testNoSameWriterRetryWhenRotationCannotStageWriter() throws Exception { + final String tableName = "TBLSAFR"; + final long commitId = 1L; + final Mutation put = LogFileTestUtil.newPut("row", 1, 1); + + ReplicationLog activeLog = logGroup.getActiveLog(); + LogFileWriter initialWriter = activeLog.getWriter(); + assertNotNull("Initial writer should not be null", initialWriter); + + // The peer DN is dead: the current writer fails on sync, and rotation cannot mint a fresh + // writer either (createNewWriter()'s header sync would also hang) -- so no pendingWriter is + // ever staged. + doThrow(new IOException("Simulated sync failure")).when(initialWriter).sync(); + doThrow(new IOException("Simulated rotation failure")).when(activeLog).createNewWriter(); + + logGroup.append(tableName, commitId, put); + logGroup.sync(); + + // Only ONE sync on the fenced writer: apply() saw no staged writer and surfaced the failure + // instead of burning attempt 2 on the same writer. + verify(initialWriter, times(1)).sync(); + assertEquals(STORE_AND_FORWARD, logGroup.getMode()); + + // RPO survival: the unsynced record must land on the SAF-mode writer. A downgrade that dropped + // currentBatch would still pass the mode + sync-count assertions above but silently lose the + // record -- the exact false-success this scenario guards against. Mirrors the append-path + // assertion in testBlockFullAppendFailureRecoveredViaReplayFailedEvent(). + LogFileWriter safWriter = logGroup.getActiveLog().getWriter(); + assertNotEquals("SAF writer must be a different instance than the fenced SYNC writer", + initialWriter, safWriter); + verify(safWriter, times(1)).append(eq(tableName), eq(commitId), any(List.class), any()); + } + /** * Tests the behavior when we fail to update the HAGroup store status when we switch to the * STORE_AND_FORWARD mode and abort @@ -1749,6 +1790,73 @@ public void testErrorRecoveryRequestsNewWriter() throws Exception { verify(newWriter, times(1)).sync(); } + /** + * PHOENIX-7984 the apply() retry re-drives rotation on every spin, so a FIRST rotation attempt + * that fails to stage a writer is retried within the {@code retryDelayMs} budget instead of + * giving up. This is the missed-retry / swallowed-request defect the addendum fixed: a single + * request-then-wait (the pre-fix shape) gives up after one failed rotation and downgrades SYNC to + * STORE_AND_FORWARD prematurely, even though a fresh writer was moments away. + *

+ * We fail the write once on the initial writer, then make the first rotation's createNewWriter() + * throw and the second succeed. Reverting awaitStagedWriter() to a single + * requestRotation()+wait() would time out after the first (failed) rotation, surface the original + * failure, and flip to SAF -- so this test fails on the pre-addendum shape while passing on the + * re-drive loop. + *

+ * Determinism: the rotation executor is single-threaded, so the failing rotation runs to + * completion (clearing the CAS gate and notifying) before the re-driven one is scheduled; the + * guarded wait on {@code rotationSignal} guarantees the retry observes the staged writer with no + * lost wakeup. {@code retryDelayMs} is raised well above two local-FS rotations so the budget + * cannot expire mid-sequence, and the scheduled tick stays at the 60s default so it never fires. + */ + @Test + public void testRetryReDrivesRotationAfterFirstRotationFails() throws Exception { + final String tableName = "TBLRDRAFRF"; + final Mutation put = LogFileTestUtil.newPut("row", 1, 1); + final long commitId = 1L; + + // Give the bounded wait plenty of room for two sequential rotations (fail then succeed) so the + // outcome turns on the re-drive, not on the timeout racing local-FS latency. + conf.setLong(ReplicationLogGroup.REPLICATION_LOG_RETRY_DELAY_MS_KEY, 5000L); + recreateLogGroup(); + + ReplicationLog activeLog = logGroup.getActiveLog(); + LogFileWriter initialWriter = activeLog.getWriter(); + assertNotNull("Initial writer should not be null", initialWriter); + + // Attempt 1 fails on the initial writer, forcing the retry into awaitStagedWriter(). + doThrow(new IOException("Simulated broken stream")).when(initialWriter).append(anyString(), + anyLong(), any(List.class), any()); + + // First rotation fails to mint a writer; second succeeds. Only the re-drive loop issues that + // second request. One failure is well under maxRotationRetries (5), so the log is not closed. + final AtomicInteger rotationAttempts = new AtomicInteger(0); + doAnswer(invocation -> { + if (rotationAttempts.incrementAndGet() == 1) { + throw new IOException("Simulated first rotation failure"); + } + return invocation.callRealMethod(); + }).when(activeLog).createNewWriter(); + + logGroup.append(tableName, commitId, put); + logGroup.sync(); + + // The re-drive scheduled a second rotation that staged a fresh writer, and the retry landed on + // it: mode stayed SYNC (no premature downgrade) and the record is on the new writer. + assertEquals("Retry must recover on the re-driven writer, not downgrade to SAF", SYNC, + logGroup.getMode()); + LogFileWriter newWriter = activeLog.getWriter(); + assertNotEquals("Should be using a fresh writer after the re-driven rotation", initialWriter, + newWriter); + assertTrue("Both rotation attempts must have run (first failed, second staged)", + rotationAttempts.get() >= 2); + verify(initialWriter, times(1)).append(eq(tableName), eq(commitId), + eq(LogFileTestUtil.cellsOf(put)), any()); + verify(newWriter, times(1)).append(eq(tableName), eq(commitId), + eq(LogFileTestUtil.cellsOf(put)), any()); + verify(newWriter, times(1)).sync(); + } + /** * Tests that an on-demand size rotation mid-interval does not suppress the next scheduled tick. * After size rotation creates a writer early, the scheduled tick still fires and creates another. @@ -2110,6 +2218,83 @@ public void testBlockFullSyncOnAppendReducesReplayOnRotation() throws Exception + newWriterAppendCount, newWriterAppendCount < id); } + /** + * S17b (append path): a record whose block-full sync fails is recovered via + * {@code replayFailedEvent}, NOT {@code replayBatch}. When {@link LogFileWriter#append} fills a + * block it syncs internally; if that durability barrier fails the append throws. Because + * {@link ReplicationLog#append} only adds the record to {@code currentBatch} AFTER the append + * returns, the failing record R is never in the batch -- so {@code replayBatch(currentBatch)} + * cannot bring it back. Recovery comes solely from {@code LogEventHandler.replayFailedEvent}, + * which re-appends the failing DATA event's own record onto the fresh SAF-mode writer. + *

+ * This exercises the real block-full path (tiny block size, as in + * {@link #testBlockFullSyncOnAppendReducesReplayOnRotation()}): failure is injected at a genuine + * block-full boundary (the append that reports a block sync), and rotation is blocked so the SYNC + * writer cannot be rescued -- forcing the SYNC->SAF transition. We assert that R was absent from + * the SYNC log's {@code currentBatch} at the moment of failure, yet still lands on the SAF + * writer. + */ + @Test + public void testBlockFullAppendFailureRecoveredViaReplayFailedEvent() throws Exception { + final String tableName = "TBLBFAF"; + // Tiny blocks so appends fill a block quickly and trigger the block-full internal sync. + conf.setLong(LogFileWriterContext.LOGFILE_BLOCK_SIZE, 200L); + recreateLogGroup(); + + ReplicationLog syncLog = logGroup.getActiveLog(); + LogFileWriter syncWriter = syncLog.getWriter(); + assertNotNull("Initial writer should not be null", syncWriter); + + // R = the commitId of the record whose block-full sync fails. Captured at the failure point. + final AtomicLong failedCommitId = new AtomicLong(-1); + // Snapshot of the SYNC log's currentBatch commitIds at the failure moment (before R is added). + final List batchAtFailure = Collections.synchronizedList(new ArrayList<>()); + + // Delegate to the real append until one crosses a block boundary (returns true = block synced); + // at that genuine boundary, simulate the peer-DN sync failure by throwing. R never reaches + // currentBatch because ReplicationLog.append() adds it only after append() returns. + doAnswer(invocation -> { + boolean blockSynced = (boolean) invocation.callRealMethod(); + if (blockSynced && failedCommitId.get() < 0) { + long commitId = invocation.getArgument(1); + failedCommitId.set(commitId); + for (ReplicationLogGroup.Record r : syncLog.getCurrentBatch()) { + batchAtFailure.add(r.commitId); + } + throw new IOException("Simulated block-full sync failure against dead peer DataNode"); + } + return blockSynced; + }).when(syncWriter).append(anyString(), anyLong(), any(List.class), any()); + + // Rotation cannot rescue the retry (createNewWriter's header sync would also hang) -> SAF. + doThrow(new IOException("Simulated rotation failure")).when(syncLog).createNewWriter(); + + // Append until the block-full boundary fires the failure and flips the mode to SAF. + long id = 1; + while (logGroup.getMode() == SYNC && id <= 50) { + logGroup.append(tableName, id, LogFileTestUtil.newPut("row_" + id, id, 2)); + id++; + } + logGroup.sync(); + + assertEquals("Block-full sync failure must switch to STORE_AND_FORWARD", STORE_AND_FORWARD, + logGroup.getMode()); + long r = failedCommitId.get(); + assertTrue("Test must have hit a real block-full boundary to inject the failure", r > 0); + + // The interesting property: R was NOT in currentBatch at failure, so replayBatch could not have + // recovered it -- only replayFailedEvent can. + assertFalse( + "Failing record R must be absent from currentBatch (added only after append succeeds)", + batchAtFailure.contains(r)); + + // Yet R still reaches the SAF-mode writer, proving replayFailedEvent re-appended it. + LogFileWriter safWriter = logGroup.getActiveLog().getWriter(); + assertNotEquals("SAF writer must be a different instance than the fenced SYNC writer", + syncWriter, safWriter); + verify(safWriter, times(1)).append(eq(tableName), eq(r), any(List.class), any()); + } + /** * Tests that the size check is not invoked from inside apply(). Pre-fix, every successful action * inside apply() ran requestRotationIfOversized(); when a rotation swapped in a new writer diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileWriterSyncTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileWriterSyncTest.java index baccf55973f..4c7b485da13 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileWriterSyncTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileWriterSyncTest.java @@ -17,7 +17,10 @@ */ package org.apache.phoenix.replication.log; +import static org.junit.Assert.fail; import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -193,6 +196,71 @@ public void testSyncEmpty() throws IOException { verify(internalOutput, times(1)).hsync(); } + /** + * S17b guard: once a {@code sync()} fails at the durability barrier, the writer is fenced. A + * subsequent {@code sync()} on the SAME writer must NOT re-enter the underlying stream and must + * NOT report success -- it must throw. + *

+ * In S17b the first {@code output.sync()} threw (dead peer DataNode) after {@code closeBlock()} + * had already advanced block state, so the original code's retry {@code sync()} found nothing to + * do and returned success -- a false ACK that cleared {@code currentBatch} and silently lost the + * record. The fix fences the writer on first failure (mirroring HDFS DFSOutputStream, whose + * stream tears itself down and rethrows on any call after a failed sync), so a same-writer retry + * fails fast instead of false-succeeding. Recovery is the higher layer's job: rotate to a fresh + * writer and replay the unsynced batch. + */ + @Test + public void testWriterFencedAfterSyncFailure() throws IOException { + Mutation m1 = LogFileTestUtil.newPut("row1", 1L, 1); + writer.append("TBL", 1L, LogFileTestUtil.cellsOf(m1)); + + // First sync fails at the durability barrier (dead peer DataNode pipeline). + doThrow(new IOException("simulated peer DataNode pipeline failure")).when(internalOutput) + .hsync(); + try { + writer.sync(); + fail("Expected first sync() to propagate the hsync failure"); + } catch (IOException expected) { + // The S17b starting condition. + } + + // Even if the underlying stream would now succeed, the fenced writer must refuse to re-drive + // it: no false success, and no second hsync on the torn-down stream. + doNothing().when(internalOutput).hsync(); + clearInvocations(internalOutput); + try { + writer.sync(); + fail("Retry sync() on a fenced writer must throw, not report false success (S17b)"); + } catch (IOException expected) { + // Correct: failure surfaced, record stays a replay candidate. + } + verify(internalOutput, never()).hsync(); + } + + /** A fenced writer also refuses further appends, so no record silently accumulates on it. */ + @Test + public void testWriterFencedRejectsAppendAfterSyncFailure() throws IOException { + Mutation m1 = LogFileTestUtil.newPut("row1", 1L, 1); + writer.append("TBL", 1L, LogFileTestUtil.cellsOf(m1)); + + doThrow(new IOException("simulated peer DataNode pipeline failure")).when(internalOutput) + .hsync(); + try { + writer.sync(); + fail("Expected first sync() to propagate the hsync failure"); + } catch (IOException expected) { + // expected + } + + Mutation m2 = LogFileTestUtil.newPut("row2", 2L, 1); + try { + writer.append("TBL", 2L, LogFileTestUtil.cellsOf(m2)); + fail("append() on a fenced writer must throw"); + } catch (IOException expected) { + // Correct: the caller must route this record to a fresh writer, not this fenced one. + } + } + @Test public void testSyncWithHflush() throws IOException { // Create a separate writer with default config (hbase.wal.hsync=false)