@Test public void testRollback() throws IOException, InterruptedException { final int rowCountOfRegionA = loadRegion(this.region_a, CF, true); final int rowCountOfRegionB = loadRegion(this.region_b, CF, true); assertTrue(rowCountOfRegionA > 0 && rowCountOfRegionB > 0); assertEquals(rowCountOfRegionA, countRows(this.region_a)); assertEquals(rowCountOfRegionB, countRows(this.region_b)); RegionMergeTransaction mt = prepareOnGoodRegions(); when(mt.createMergedRegionFromMerges(region_a, region_b, mt.getMergedRegionInfo())).thenThrow( new MockedFailedMergedRegionCreation()); boolean expectedException = false; TEST_UTIL.getConfiguration().setInt(HConstants.REGIONSERVER_PORT, 0); CoordinatedStateManager cp = CoordinatedStateManagerFactory.getCoordinatedStateManager( TEST_UTIL.getConfiguration()); Server mockServer = new HRegionServer(TEST_UTIL.getConfiguration(), cp); try { mt.execute(mockServer, null); } catch (MockedFailedMergedRegionCreation e) { expectedException = true; } assertTrue(expectedException); assertTrue(mt.rollback(null, null)); int rowCountOfRegionA2 = countRows(this.region_a); assertEquals(rowCountOfRegionA, rowCountOfRegionA2); int rowCountOfRegionB2 = countRows(this.region_b); assertEquals(rowCountOfRegionB, rowCountOfRegionB2); assertTrue(!this.fs.exists(HRegion.getRegionDir(this.testdir, mt.getMergedRegionInfo()))); assertTrue(!this.region_a.lock.writeLock().isHeldByCurrentThread()); assertTrue(!this.region_b.lock.writeLock().isHeldByCurrentThread()); assertTrue(mt.prepare(null)); HRegion mergedRegion = mt.execute(mockServer, null); try { int mergedRegionRowCount = countRows(mergedRegion); assertEquals((rowCountOfRegionA + rowCountOfRegionB), mergedRegionRowCount); } finally { HRegion.closeHRegion(mergedRegion); } assertTrue(!this.region_a.lock.writeLock().isHeldByCurrentThread()); assertTrue(!this.region_b.lock.writeLock().isHeldByCurrentThread()); }
@SuppressWarnings("deprecation") public boolean rollback(final Server server, final RegionServerServices services) throws IOException { assert this.mergedRegionInfo != null; if (rsCoprocessorHost != null) { rsCoprocessorHost.preRollBackMerge(this.region_a, this.region_b); } boolean result = true; ListIterator<JournalEntry> iterator = this.journal .listIterator(this.journal.size()); while (iterator.hasPrevious()) { JournalEntry je = iterator.previous(); switch (je) { case SET_MERGING_IN_ZK: if (useZKAndZKIsSet(server)) { cleanZK(server, this.mergedRegionInfo); } else if (services != null && !useZKForAssignment && !services.reportRegionTransition(TransitionCode.MERGE_REVERTED, mergedRegionInfo, region_a.getRegionInfo(), region_b.getRegionInfo())) { return false; } break; case CREATED_MERGE_DIR: this.region_a.writestate.writesEnabled = true; this.region_b.writestate.writesEnabled = true; this.region_a.getRegionFileSystem().cleanupMergesDir(); break; case CLOSED_REGION_A: try { this.region_a.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_a.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_A: if (services != null) services.addToOnlineRegions(this.region_a); break; case CLOSED_REGION_B: try { this.region_b.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_b.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_B: if (services != null) services.addToOnlineRegions(this.region_b); break; case STARTED_MERGED_REGION_CREATION: this.region_a.getRegionFileSystem().cleanupMergedRegion( this.mergedRegionInfo); break; case PONR: return false; default: throw new RuntimeException("Unhandled journal entry: " + je); } } if (rsCoprocessorHost != null) { rsCoprocessorHost.postRollBackMerge(this.region_a, this.region_b); } return result; }
RegionMergeTransaction { @SuppressWarnings("deprecation") public boolean rollback(final Server server, final RegionServerServices services) throws IOException { assert this.mergedRegionInfo != null; if (rsCoprocessorHost != null) { rsCoprocessorHost.preRollBackMerge(this.region_a, this.region_b); } boolean result = true; ListIterator<JournalEntry> iterator = this.journal .listIterator(this.journal.size()); while (iterator.hasPrevious()) { JournalEntry je = iterator.previous(); switch (je) { case SET_MERGING_IN_ZK: if (useZKAndZKIsSet(server)) { cleanZK(server, this.mergedRegionInfo); } else if (services != null && !useZKForAssignment && !services.reportRegionTransition(TransitionCode.MERGE_REVERTED, mergedRegionInfo, region_a.getRegionInfo(), region_b.getRegionInfo())) { return false; } break; case CREATED_MERGE_DIR: this.region_a.writestate.writesEnabled = true; this.region_b.writestate.writesEnabled = true; this.region_a.getRegionFileSystem().cleanupMergesDir(); break; case CLOSED_REGION_A: try { this.region_a.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_a.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_A: if (services != null) services.addToOnlineRegions(this.region_a); break; case CLOSED_REGION_B: try { this.region_b.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_b.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_B: if (services != null) services.addToOnlineRegions(this.region_b); break; case STARTED_MERGED_REGION_CREATION: this.region_a.getRegionFileSystem().cleanupMergedRegion( this.mergedRegionInfo); break; case PONR: return false; default: throw new RuntimeException("Unhandled journal entry: " + je); } } if (rsCoprocessorHost != null) { rsCoprocessorHost.postRollBackMerge(this.region_a, this.region_b); } return result; } }
RegionMergeTransaction { @SuppressWarnings("deprecation") public boolean rollback(final Server server, final RegionServerServices services) throws IOException { assert this.mergedRegionInfo != null; if (rsCoprocessorHost != null) { rsCoprocessorHost.preRollBackMerge(this.region_a, this.region_b); } boolean result = true; ListIterator<JournalEntry> iterator = this.journal .listIterator(this.journal.size()); while (iterator.hasPrevious()) { JournalEntry je = iterator.previous(); switch (je) { case SET_MERGING_IN_ZK: if (useZKAndZKIsSet(server)) { cleanZK(server, this.mergedRegionInfo); } else if (services != null && !useZKForAssignment && !services.reportRegionTransition(TransitionCode.MERGE_REVERTED, mergedRegionInfo, region_a.getRegionInfo(), region_b.getRegionInfo())) { return false; } break; case CREATED_MERGE_DIR: this.region_a.writestate.writesEnabled = true; this.region_b.writestate.writesEnabled = true; this.region_a.getRegionFileSystem().cleanupMergesDir(); break; case CLOSED_REGION_A: try { this.region_a.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_a.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_A: if (services != null) services.addToOnlineRegions(this.region_a); break; case CLOSED_REGION_B: try { this.region_b.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_b.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_B: if (services != null) services.addToOnlineRegions(this.region_b); break; case STARTED_MERGED_REGION_CREATION: this.region_a.getRegionFileSystem().cleanupMergedRegion( this.mergedRegionInfo); break; case PONR: return false; default: throw new RuntimeException("Unhandled journal entry: " + je); } } if (rsCoprocessorHost != null) { rsCoprocessorHost.postRollBackMerge(this.region_a, this.region_b); } return result; } RegionMergeTransaction(final HRegion a, final HRegion b,
      final boolean forcible); }
RegionMergeTransaction { @SuppressWarnings("deprecation") public boolean rollback(final Server server, final RegionServerServices services) throws IOException { assert this.mergedRegionInfo != null; if (rsCoprocessorHost != null) { rsCoprocessorHost.preRollBackMerge(this.region_a, this.region_b); } boolean result = true; ListIterator<JournalEntry> iterator = this.journal .listIterator(this.journal.size()); while (iterator.hasPrevious()) { JournalEntry je = iterator.previous(); switch (je) { case SET_MERGING_IN_ZK: if (useZKAndZKIsSet(server)) { cleanZK(server, this.mergedRegionInfo); } else if (services != null && !useZKForAssignment && !services.reportRegionTransition(TransitionCode.MERGE_REVERTED, mergedRegionInfo, region_a.getRegionInfo(), region_b.getRegionInfo())) { return false; } break; case CREATED_MERGE_DIR: this.region_a.writestate.writesEnabled = true; this.region_b.writestate.writesEnabled = true; this.region_a.getRegionFileSystem().cleanupMergesDir(); break; case CLOSED_REGION_A: try { this.region_a.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_a.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_A: if (services != null) services.addToOnlineRegions(this.region_a); break; case CLOSED_REGION_B: try { this.region_b.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_b.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_B: if (services != null) services.addToOnlineRegions(this.region_b); break; case STARTED_MERGED_REGION_CREATION: this.region_a.getRegionFileSystem().cleanupMergedRegion( this.mergedRegionInfo); break; case PONR: return false; default: throw new RuntimeException("Unhandled journal entry: " + je); } } if (rsCoprocessorHost != null) { rsCoprocessorHost.postRollBackMerge(this.region_a, this.region_b); } return result; } RegionMergeTransaction(final HRegion a, final HRegion b,
      final boolean forcible); boolean prepare(final RegionServerServices services); HRegion execute(final Server server,
      final RegionServerServices services); HRegion stepsAfterPONR(final Server server, final RegionServerServices services,
      HRegion mergedRegion); void prepareMutationsForMerge(HRegionInfo mergedRegion, HRegionInfo regionA,
      HRegionInfo regionB, ServerName serverName, List<Mutation> mutations); @SuppressWarnings("deprecation") Put addLocation(final Put p, final ServerName sn, long openSeqNum); HRegion stepsBeforePONR(final Server server, final RegionServerServices services,
      boolean testing); static HRegionInfo getMergedRegionInfo(final HRegionInfo a,
      final HRegionInfo b); @SuppressWarnings("deprecation") boolean rollback(final Server server,
      final RegionServerServices services); static void createNodeMerging(final ZooKeeperWatcher zkw, final HRegionInfo region,
      final ServerName serverName, final HRegionInfo a,
      final HRegionInfo b); static int transitionMergingNode(ZooKeeperWatcher zkw,
      HRegionInfo merged, HRegionInfo a, HRegionInfo b, ServerName serverName,
      final int znodeVersion, final EventType beginState,
      final EventType endState); }
RegionMergeTransaction { @SuppressWarnings("deprecation") public boolean rollback(final Server server, final RegionServerServices services) throws IOException { assert this.mergedRegionInfo != null; if (rsCoprocessorHost != null) { rsCoprocessorHost.preRollBackMerge(this.region_a, this.region_b); } boolean result = true; ListIterator<JournalEntry> iterator = this.journal .listIterator(this.journal.size()); while (iterator.hasPrevious()) { JournalEntry je = iterator.previous(); switch (je) { case SET_MERGING_IN_ZK: if (useZKAndZKIsSet(server)) { cleanZK(server, this.mergedRegionInfo); } else if (services != null && !useZKForAssignment && !services.reportRegionTransition(TransitionCode.MERGE_REVERTED, mergedRegionInfo, region_a.getRegionInfo(), region_b.getRegionInfo())) { return false; } break; case CREATED_MERGE_DIR: this.region_a.writestate.writesEnabled = true; this.region_b.writestate.writesEnabled = true; this.region_a.getRegionFileSystem().cleanupMergesDir(); break; case CLOSED_REGION_A: try { this.region_a.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_a.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_A: if (services != null) services.addToOnlineRegions(this.region_a); break; case CLOSED_REGION_B: try { this.region_b.initialize(); } catch (IOException e) { LOG.error("Failed rollbacking CLOSED_REGION_A of region " + this.region_b.getRegionNameAsString(), e); throw new RuntimeException(e); } break; case OFFLINED_REGION_B: if (services != null) services.addToOnlineRegions(this.region_b); break; case STARTED_MERGED_REGION_CREATION: this.region_a.getRegionFileSystem().cleanupMergedRegion( this.mergedRegionInfo); break; case PONR: return false; default: throw new RuntimeException("Unhandled journal entry: " + je); } } if (rsCoprocessorHost != null) { rsCoprocessorHost.postRollBackMerge(this.region_a, this.region_b); } return result; } RegionMergeTransaction(final HRegion a, final HRegion b,
      final boolean forcible); boolean prepare(final RegionServerServices services); HRegion execute(final Server server,
      final RegionServerServices services); HRegion stepsAfterPONR(final Server server, final RegionServerServices services,
      HRegion mergedRegion); void prepareMutationsForMerge(HRegionInfo mergedRegion, HRegionInfo regionA,
      HRegionInfo regionB, ServerName serverName, List<Mutation> mutations); @SuppressWarnings("deprecation") Put addLocation(final Put p, final ServerName sn, long openSeqNum); HRegion stepsBeforePONR(final Server server, final RegionServerServices services,
      boolean testing); static HRegionInfo getMergedRegionInfo(final HRegionInfo a,
      final HRegionInfo b); @SuppressWarnings("deprecation") boolean rollback(final Server server,
      final RegionServerServices services); static void createNodeMerging(final ZooKeeperWatcher zkw, final HRegionInfo region,
      final ServerName serverName, final HRegionInfo a,
      final HRegionInfo b); static int transitionMergingNode(ZooKeeperWatcher zkw,
      HRegionInfo merged, HRegionInfo a, HRegionInfo b, ServerName serverName,
      final int znodeVersion, final EventType beginState,
      final EventType endState);  }