001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.mapreduce;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertTrue;
022import static org.junit.jupiter.api.Assertions.fail;
023
024import java.util.Arrays;
025import java.util.function.BooleanSupplier;
026import org.apache.commons.lang3.ArrayUtils;
027import org.apache.hadoop.conf.Configuration;
028import org.apache.hadoop.fs.FileSystem;
029import org.apache.hadoop.fs.Path;
030import org.apache.hadoop.hbase.Cell;
031import org.apache.hadoop.hbase.CellUtil;
032import org.apache.hadoop.hbase.HBaseTestingUtil;
033import org.apache.hadoop.hbase.TableName;
034import org.apache.hadoop.hbase.client.Put;
035import org.apache.hadoop.hbase.client.Result;
036import org.apache.hadoop.hbase.client.ResultScanner;
037import org.apache.hadoop.hbase.client.Scan;
038import org.apache.hadoop.hbase.client.Table;
039import org.apache.hadoop.hbase.mapreduce.SyncTable.SyncMapper.Counter;
040import org.apache.hadoop.hbase.testclassification.LargeTests;
041import org.apache.hadoop.hbase.testclassification.MapReduceTests;
042import org.apache.hadoop.hbase.util.Bytes;
043import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
044import org.apache.hadoop.mapreduce.Counters;
045import org.junit.jupiter.api.AfterAll;
046import org.junit.jupiter.api.BeforeAll;
047import org.junit.jupiter.api.Tag;
048import org.junit.jupiter.api.Test;
049import org.junit.jupiter.api.TestInfo;
050import org.slf4j.Logger;
051import org.slf4j.LoggerFactory;
052
053/**
054 * Basic test for the SyncTable M/R tool
055 */
056@Tag(MapReduceTests.TAG)
057@Tag(LargeTests.TAG)
058public class TestSyncTable {
059
060  private static final Logger LOG = LoggerFactory.getLogger(TestSyncTable.class);
061
062  private static final HBaseTestingUtil UTIL1 = new HBaseTestingUtil();
063
064  private static final HBaseTestingUtil UTIL2 = new HBaseTestingUtil();
065
066  @BeforeAll
067  public static void beforeClass() throws Exception {
068    UTIL1.startMiniCluster(3);
069    UTIL2.startMiniCluster(3);
070  }
071
072  @AfterAll
073  public static void afterClass() throws Exception {
074    UTIL2.shutdownMiniCluster();
075    UTIL1.shutdownMiniCluster();
076  }
077
078  private static byte[][] generateSplits(int numRows, int numRegions) {
079    byte[][] splitRows = new byte[numRegions - 1][];
080    for (int i = 1; i < numRegions; i++) {
081      splitRows[i - 1] = Bytes.toBytes(numRows * i / numRegions);
082    }
083    return splitRows;
084  }
085
086  private void testSyncTable(TestInfo testInfo, HBaseTestingUtil source, HBaseTestingUtil target,
087    String... options) throws Exception {
088    final TableName sourceTableName =
089      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_source");
090    final TableName targetTableName =
091      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_target");
092    Path testDir = source.getDataTestDirOnTestFS(testInfo.getTestMethod().get().getName());
093
094    writeTestData(source, sourceTableName, target, targetTableName);
095    hashSourceTable(source, sourceTableName, testDir);
096    Counters syncCounters =
097      syncTables(target.getConfiguration(), sourceTableName, targetTableName, testDir, options);
098    assertEqualTables(90, source, sourceTableName, target, targetTableName, false);
099
100    assertEquals(60, syncCounters.findCounter(Counter.ROWSWITHDIFFS).getValue());
101    assertEquals(10, syncCounters.findCounter(Counter.SOURCEMISSINGROWS).getValue());
102    assertEquals(10, syncCounters.findCounter(Counter.TARGETMISSINGROWS).getValue());
103    assertEquals(50, syncCounters.findCounter(Counter.SOURCEMISSINGCELLS).getValue());
104    assertEquals(50, syncCounters.findCounter(Counter.TARGETMISSINGCELLS).getValue());
105    assertEquals(20, syncCounters.findCounter(Counter.DIFFERENTCELLVALUES).getValue());
106
107    source.deleteTable(sourceTableName);
108    target.deleteTable(targetTableName);
109  }
110
111  @Test
112  public void testSyncTable(TestInfo testInfo) throws Exception {
113    testSyncTable(testInfo, UTIL1, UTIL1);
114  }
115
116  @Test
117  public void testSyncTableToPeerCluster(TestInfo testInfo) throws Exception {
118    testSyncTable(testInfo, UTIL1, UTIL2, "--sourceuri=" + UTIL1.getRpcConnnectionURI());
119  }
120
121  @Test
122  public void testSyncTableFromSourceToPeerCluster(TestInfo testInfo) throws Exception {
123    testSyncTable(testInfo, UTIL2, UTIL1, "--sourceuri=" + UTIL2.getRpcConnnectionURI(),
124      "--targeturi=" + UTIL1.getZkConnectionURI());
125  }
126
127  @Test
128  public void testSyncTableFromSourceToPeerClusterWithClusterKey(TestInfo testInfo)
129    throws Exception {
130    testSyncTable(testInfo, UTIL2, UTIL1, "--sourcezkcluster=" + UTIL2.getClusterKey(),
131      "--targetzkcluster=" + UTIL1.getClusterKey());
132  }
133
134  @Test
135  public void testSyncTableDoDeletesFalse(TestInfo testInfo) throws Exception {
136    final TableName sourceTableName =
137      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_source");
138    final TableName targetTableName =
139      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_target");
140    Path testDir = UTIL1.getDataTestDirOnTestFS(testInfo.getTestMethod().get().getName());
141
142    writeTestData(UTIL1, sourceTableName, UTIL1, targetTableName);
143    hashSourceTable(UTIL1, sourceTableName, testDir);
144    Counters syncCounters = syncTables(UTIL1.getConfiguration(), sourceTableName, targetTableName,
145      testDir, "--doDeletes=false");
146    assertTargetDoDeletesFalse(100, UTIL1, sourceTableName, UTIL1, targetTableName);
147
148    assertEquals(60, syncCounters.findCounter(Counter.ROWSWITHDIFFS).getValue());
149    assertEquals(10, syncCounters.findCounter(Counter.SOURCEMISSINGROWS).getValue());
150    assertEquals(10, syncCounters.findCounter(Counter.TARGETMISSINGROWS).getValue());
151    assertEquals(50, syncCounters.findCounter(Counter.SOURCEMISSINGCELLS).getValue());
152    assertEquals(50, syncCounters.findCounter(Counter.TARGETMISSINGCELLS).getValue());
153    assertEquals(20, syncCounters.findCounter(Counter.DIFFERENTCELLVALUES).getValue());
154
155    UTIL1.deleteTable(sourceTableName);
156    UTIL1.deleteTable(targetTableName);
157  }
158
159  @Test
160  public void testSyncTableDoPutsFalse(TestInfo testInfo) throws Exception {
161    final TableName sourceTableName =
162      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_source");
163    final TableName targetTableName =
164      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_target");
165    Path testDir = UTIL2.getDataTestDirOnTestFS(testInfo.getTestMethod().get().getName());
166
167    writeTestData(UTIL2, sourceTableName, UTIL2, targetTableName);
168    hashSourceTable(UTIL2, sourceTableName, testDir);
169    Counters syncCounters = syncTables(UTIL2.getConfiguration(), sourceTableName, targetTableName,
170      testDir, "--doPuts=false");
171    assertTargetDoPutsFalse(70, UTIL2, sourceTableName, UTIL2, targetTableName);
172
173    assertEquals(60, syncCounters.findCounter(Counter.ROWSWITHDIFFS).getValue());
174    assertEquals(10, syncCounters.findCounter(Counter.SOURCEMISSINGROWS).getValue());
175    assertEquals(10, syncCounters.findCounter(Counter.TARGETMISSINGROWS).getValue());
176    assertEquals(50, syncCounters.findCounter(Counter.SOURCEMISSINGCELLS).getValue());
177    assertEquals(50, syncCounters.findCounter(Counter.TARGETMISSINGCELLS).getValue());
178    assertEquals(20, syncCounters.findCounter(Counter.DIFFERENTCELLVALUES).getValue());
179
180    UTIL2.deleteTable(sourceTableName);
181    UTIL2.deleteTable(targetTableName);
182  }
183
184  @Test
185  public void testSyncTableWithSha256(TestInfo testInfo) throws Exception {
186    final TableName sourceTableName =
187      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_source");
188    final TableName targetTableName =
189      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_target");
190    Path testDir = UTIL1.getDataTestDirOnTestFS(testInfo.getTestMethod().get().getName());
191
192    writeTestData(UTIL1, sourceTableName, UTIL1, targetTableName);
193    hashSourceTable(UTIL1, sourceTableName, testDir, "--hashAlgorithm=SHA-256");
194
195    HashTable.TableHash tableHash =
196      HashTable.TableHash.read(UTIL1.getTestFileSystem().getConf(), testDir);
197    assertEquals("SHA-256", tableHash.hashAlgorithm,
198      "manifest must carry the algorithm so SyncTable can match the source-side digest");
199
200    Counters syncCounters =
201      syncTables(UTIL1.getConfiguration(), sourceTableName, targetTableName, testDir);
202    assertEqualTables(90, UTIL1, sourceTableName, UTIL1, targetTableName, false);
203
204    assertEquals(60, syncCounters.findCounter(Counter.ROWSWITHDIFFS).getValue());
205    assertEquals(10, syncCounters.findCounter(Counter.SOURCEMISSINGROWS).getValue());
206    assertEquals(10, syncCounters.findCounter(Counter.TARGETMISSINGROWS).getValue());
207    assertEquals(50, syncCounters.findCounter(Counter.SOURCEMISSINGCELLS).getValue());
208    assertEquals(50, syncCounters.findCounter(Counter.TARGETMISSINGCELLS).getValue());
209    assertEquals(20, syncCounters.findCounter(Counter.DIFFERENTCELLVALUES).getValue());
210
211    UTIL1.deleteTable(sourceTableName);
212    UTIL1.deleteTable(targetTableName);
213  }
214
215  @Test
216  public void testSyncTableIgnoreTimestampsTrue(TestInfo testInfo) throws Exception {
217    final TableName sourceTableName =
218      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_source");
219    final TableName targetTableName =
220      TableName.valueOf(testInfo.getTestMethod().get().getName() + "_target");
221    Path testDir = UTIL1.getDataTestDirOnTestFS(testInfo.getTestMethod().get().getName());
222    long current = EnvironmentEdgeManager.currentTime();
223    writeTestData(UTIL1, sourceTableName, UTIL2, targetTableName, current - 1000, current);
224    hashSourceTable(UTIL1, sourceTableName, testDir, "--ignoreTimestamps=true");
225    Counters syncCounters = syncTables(UTIL2.getConfiguration(), sourceTableName, targetTableName,
226      testDir, "--ignoreTimestamps=true", "--sourceuri=" + UTIL1.getRpcConnnectionURI());
227    assertEqualTables(90, UTIL1, sourceTableName, UTIL2, targetTableName, true);
228
229    assertEquals(50, syncCounters.findCounter(Counter.ROWSWITHDIFFS).getValue());
230    assertEquals(10, syncCounters.findCounter(Counter.SOURCEMISSINGROWS).getValue());
231    assertEquals(10, syncCounters.findCounter(Counter.TARGETMISSINGROWS).getValue());
232    assertEquals(30, syncCounters.findCounter(Counter.SOURCEMISSINGCELLS).getValue());
233    assertEquals(30, syncCounters.findCounter(Counter.TARGETMISSINGCELLS).getValue());
234    assertEquals(20, syncCounters.findCounter(Counter.DIFFERENTCELLVALUES).getValue());
235
236    UTIL1.deleteTable(sourceTableName);
237    UTIL2.deleteTable(targetTableName);
238  }
239
240  private void assertCellEquals(Cell sourceCell, Cell targetCell, BooleanSupplier checkTimestamp) {
241    assertTrue(CellUtil.matchingRows(sourceCell, targetCell),
242      "Rows don't match, source: " + sourceCell + ", target: " + targetCell);
243    assertTrue(CellUtil.matchingFamily(sourceCell, targetCell),
244      "Families don't match, source: " + sourceCell + ", target: " + targetCell);
245    assertTrue(CellUtil.matchingQualifier(sourceCell, targetCell),
246      "Qualifiers don't match, source: " + sourceCell + ", target: " + targetCell);
247    if (checkTimestamp.getAsBoolean()) {
248      assertTrue(CellUtil.matchingTimestamp(sourceCell, targetCell),
249        "Timestamps don't match, source: " + sourceCell + ", target: " + targetCell);
250    }
251    assertTrue(CellUtil.matchingValue(sourceCell, targetCell),
252      "Values don't match, source: " + sourceCell + ", target: " + targetCell);
253  }
254
255  private void assertEqualTables(int expectedRows, HBaseTestingUtil sourceCluster,
256    TableName sourceTableName, HBaseTestingUtil targetCluster, TableName targetTableName,
257    boolean ignoreTimestamps) throws Exception {
258    try (Table sourceTable = sourceCluster.getConnection().getTable(sourceTableName);
259      Table targetTable = targetCluster.getConnection().getTable(targetTableName);
260      ResultScanner sourceScanner = sourceTable.getScanner(new Scan());
261      ResultScanner targetScanner = targetTable.getScanner(new Scan())) {
262      for (int i = 0; i < expectedRows; i++) {
263        Result sourceRow = sourceScanner.next();
264        Result targetRow = targetScanner.next();
265
266        LOG.debug("SOURCE row: " + (sourceRow == null ? "null" : Bytes.toInt(sourceRow.getRow()))
267          + " cells:" + sourceRow);
268        LOG.debug("TARGET row: " + (targetRow == null ? "null" : Bytes.toInt(targetRow.getRow()))
269          + " cells:" + targetRow);
270
271        if (sourceRow == null) {
272          fail("Expected " + expectedRows + " source rows but only found " + i);
273        }
274        if (targetRow == null) {
275          fail("Expected " + expectedRows + " target rows but only found " + i);
276        }
277        Cell[] sourceCells = sourceRow.rawCells();
278        Cell[] targetCells = targetRow.rawCells();
279        if (sourceCells.length != targetCells.length) {
280          LOG.debug("Source cells: " + Arrays.toString(sourceCells));
281          LOG.debug("Target cells: " + Arrays.toString(targetCells));
282          fail("Row " + Bytes.toInt(sourceRow.getRow()) + " has " + sourceCells.length
283            + " cells in source table but " + targetCells.length + " cells in target table");
284        }
285        for (int j = 0; j < sourceCells.length; j++) {
286          Cell sourceCell = sourceCells[j];
287          Cell targetCell = targetCells[j];
288          assertCellEquals(sourceCell, targetCell, () -> !ignoreTimestamps);
289        }
290      }
291      Result sourceRow = sourceScanner.next();
292      if (sourceRow != null) {
293        fail("Source table has more than " + expectedRows + " rows.  Next row: "
294          + Bytes.toInt(sourceRow.getRow()));
295      }
296      Result targetRow = targetScanner.next();
297      if (targetRow != null) {
298        fail("Target table has more than " + expectedRows + " rows.  Next row: "
299          + Bytes.toInt(targetRow.getRow()));
300      }
301    }
302  }
303
304  private void assertTargetDoDeletesFalse(int expectedRows, HBaseTestingUtil sourceCluster,
305    TableName sourceTableName, HBaseTestingUtil targetCluster, TableName targetTableName)
306    throws Exception {
307    try (Table sourceTable = sourceCluster.getConnection().getTable(sourceTableName);
308      Table targetTable = targetCluster.getConnection().getTable(targetTableName);
309
310      ResultScanner sourceScanner = sourceTable.getScanner(new Scan());
311      ResultScanner targetScanner = targetTable.getScanner(new Scan())) {
312      Result targetRow = targetScanner.next();
313      Result sourceRow = sourceScanner.next();
314      int rowsCount = 0;
315      while (targetRow != null) {
316        rowsCount++;
317        // only compares values for existing rows, skipping rows existing on
318        // target only that were not deleted given --doDeletes=false
319        if (Bytes.toInt(sourceRow.getRow()) != Bytes.toInt(targetRow.getRow())) {
320          targetRow = targetScanner.next();
321          continue;
322        }
323
324        LOG.debug("SOURCE row: " + (sourceRow == null ? "null" : Bytes.toInt(sourceRow.getRow()))
325          + " cells:" + sourceRow);
326        LOG.debug("TARGET row: " + (targetRow == null ? "null" : Bytes.toInt(targetRow.getRow()))
327          + " cells:" + targetRow);
328
329        Cell[] sourceCells = sourceRow.rawCells();
330        Cell[] targetCells = targetRow.rawCells();
331        int targetRowKey = Bytes.toInt(targetRow.getRow());
332        if (targetRowKey >= 70 && targetRowKey < 80) {
333          if (sourceCells.length == targetCells.length) {
334            LOG.debug("Source cells: " + Arrays.toString(sourceCells));
335            LOG.debug("Target cells: " + Arrays.toString(targetCells));
336            fail("Row " + targetRowKey + " should have more cells in " + "target than in source");
337          }
338
339        } else {
340          if (sourceCells.length != targetCells.length) {
341            LOG.debug("Source cells: " + Arrays.toString(sourceCells));
342            LOG.debug("Target cells: " + Arrays.toString(targetCells));
343            fail("Row " + Bytes.toInt(sourceRow.getRow()) + " has " + sourceCells.length
344              + " cells in source table but " + targetCells.length + " cells in target table");
345          }
346        }
347        for (int j = 0; j < sourceCells.length; j++) {
348          Cell sourceCell = sourceCells[j];
349          Cell targetCell = targetCells[j];
350          assertCellEquals(sourceCell, targetCell, () -> targetRowKey < 80 && targetRowKey >= 90);
351        }
352        targetRow = targetScanner.next();
353        sourceRow = sourceScanner.next();
354      }
355      assertEquals(expectedRows, rowsCount, "Target expected rows does not match.");
356    }
357  }
358
359  private void assertTargetDoPutsFalse(int expectedRows, HBaseTestingUtil sourceCluster,
360    TableName sourceTableName, HBaseTestingUtil targetCluster, TableName targetTableName)
361    throws Exception {
362    try (Table sourceTable = sourceCluster.getConnection().getTable(sourceTableName);
363      Table targetTable = targetCluster.getConnection().getTable(targetTableName);
364      ResultScanner sourceScanner = sourceTable.getScanner(new Scan());
365      ResultScanner targetScanner = targetTable.getScanner(new Scan())) {
366      Result targetRow = targetScanner.next();
367      Result sourceRow = sourceScanner.next();
368      int rowsCount = 0;
369
370      while (targetRow != null) {
371        // only compares values for existing rows, skipping rows existing on
372        // source only that were not added to target given --doPuts=false
373        if (Bytes.toInt(sourceRow.getRow()) != Bytes.toInt(targetRow.getRow())) {
374          sourceRow = sourceScanner.next();
375          continue;
376        }
377
378        LOG.debug("SOURCE row: " + (sourceRow == null ? "null" : Bytes.toInt(sourceRow.getRow()))
379          + " cells:" + sourceRow);
380        LOG.debug("TARGET row: " + (targetRow == null ? "null" : Bytes.toInt(targetRow.getRow()))
381          + " cells:" + targetRow);
382
383        LOG.debug("rowsCount: " + rowsCount);
384
385        Cell[] sourceCells = sourceRow.rawCells();
386        Cell[] targetCells = targetRow.rawCells();
387        int targetRowKey = Bytes.toInt(targetRow.getRow());
388        if (targetRowKey >= 40 && targetRowKey < 60) {
389          LOG.debug("Source cells: " + Arrays.toString(sourceCells));
390          LOG.debug("Target cells: " + Arrays.toString(targetCells));
391          fail("There shouldn't exist any rows between 40 and 60, since "
392            + "Puts are disabled and Deletes are enabled.");
393        } else if (targetRowKey >= 60 && targetRowKey < 70) {
394          if (sourceCells.length == targetCells.length) {
395            LOG.debug("Source cells: " + Arrays.toString(sourceCells));
396            LOG.debug("Target cells: " + Arrays.toString(targetCells));
397            fail(
398              "Row " + Bytes.toInt(sourceRow.getRow()) + " shouldn't have same number of cells.");
399          }
400        } else if (targetRowKey >= 80 && targetRowKey < 90) {
401          LOG.debug("Source cells: " + Arrays.toString(sourceCells));
402          LOG.debug("Target cells: " + Arrays.toString(targetCells));
403          fail("There should be no rows between 80 and 90 on target, as "
404            + "these had different timestamps and should had been deleted.");
405        } else if (targetRowKey >= 90 && targetRowKey < 100) {
406          for (int j = 0; j < sourceCells.length; j++) {
407            Cell sourceCell = sourceCells[j];
408            Cell targetCell = targetCells[j];
409            if (CellUtil.matchingValue(sourceCell, targetCell)) {
410              fail("Cells values should not match for rows between " + "90 and 100. Target row id: "
411                + Bytes.toInt(targetRow.getRow()));
412            }
413          }
414        } else {
415          for (int j = 0; j < sourceCells.length; j++) {
416            Cell sourceCell = sourceCells[j];
417            Cell targetCell = targetCells[j];
418            assertCellEquals(sourceCell, targetCell, () -> true);
419          }
420        }
421        rowsCount++;
422        targetRow = targetScanner.next();
423        sourceRow = sourceScanner.next();
424      }
425      assertEquals(expectedRows, rowsCount, "Target expected rows does not match.");
426    }
427  }
428
429  private Counters syncTables(Configuration conf, TableName sourceTableName,
430    TableName targetTableName, Path testDir, String... options) throws Exception {
431    SyncTable syncTable = new SyncTable(conf);
432    String[] args = Arrays.copyOf(options, options.length + 3);
433    args[options.length] = testDir.toString();
434    args[options.length + 1] = sourceTableName.getNameAsString();
435    args[options.length + 2] = targetTableName.getNameAsString();
436    int code = syncTable.run(args);
437    assertEquals(0, code, "sync table job failed");
438
439    LOG.info("Sync tables completed");
440    return syncTable.counters;
441  }
442
443  private void hashSourceTable(HBaseTestingUtil sourceCluster, TableName sourceTableName,
444    Path testDir, String... options) throws Exception {
445    int numHashFiles = 3;
446    long batchSize = 100; // should be 2 batches per region
447    int scanBatch = 1;
448    HashTable hashTable = new HashTable(sourceCluster.getConfiguration());
449    String[] args = Arrays.copyOf(options, options.length + 5);
450    args[options.length] = "--batchsize=" + batchSize;
451    args[options.length + 1] = "--numhashfiles=" + numHashFiles;
452    args[options.length + 2] = "--scanbatch=" + scanBatch;
453    args[options.length + 3] = sourceTableName.getNameAsString();
454    args[options.length + 4] = testDir.toString();
455    int code = hashTable.run(args);
456    assertEquals(0, code, "hash table job failed");
457
458    FileSystem fs = sourceCluster.getTestFileSystem();
459
460    HashTable.TableHash tableHash = HashTable.TableHash.read(fs.getConf(), testDir);
461    assertEquals(sourceTableName.getNameAsString(), tableHash.tableName);
462    assertEquals(batchSize, tableHash.batchSize);
463    assertEquals(numHashFiles, tableHash.numHashFiles);
464    assertEquals(numHashFiles - 1, tableHash.partitions.size());
465
466    LOG.info("Hash table completed");
467  }
468
469  private void writeTestData(HBaseTestingUtil sourceCluster, TableName sourceTableName,
470    HBaseTestingUtil targetCluster, TableName targetTableName, long... timestamps)
471    throws Exception {
472    final byte[] family = Bytes.toBytes("family");
473    final byte[] column1 = Bytes.toBytes("c1");
474    final byte[] column2 = Bytes.toBytes("c2");
475    final byte[] value1 = Bytes.toBytes("val1");
476    final byte[] value2 = Bytes.toBytes("val2");
477    final byte[] value3 = Bytes.toBytes("val3");
478
479    int numRows = 100;
480    int sourceRegions = 10;
481    int targetRegions = 6;
482    if (ArrayUtils.isEmpty(timestamps)) {
483      long current = EnvironmentEdgeManager.currentTime();
484      timestamps = new long[] { current, current };
485    }
486
487    try (
488      Table sourceTable =
489        sourceCluster.createTable(sourceTableName, family, generateSplits(numRows, sourceRegions));
490      Table targetTable = targetCluster.createTable(targetTableName, family,
491        generateSplits(numRows, targetRegions))) {
492
493      int rowIndex = 0;
494      // a bunch of identical rows
495      for (; rowIndex < 40; rowIndex++) {
496        Put sourcePut = new Put(Bytes.toBytes(rowIndex));
497        sourcePut.addColumn(family, column1, timestamps[0], value1);
498        sourcePut.addColumn(family, column2, timestamps[0], value2);
499        sourceTable.put(sourcePut);
500
501        Put targetPut = new Put(Bytes.toBytes(rowIndex));
502        targetPut.addColumn(family, column1, timestamps[1], value1);
503        targetPut.addColumn(family, column2, timestamps[1], value2);
504        targetTable.put(targetPut);
505      }
506      // some rows only in the source table
507      // ROWSWITHDIFFS: 10
508      // TARGETMISSINGROWS: 10
509      // TARGETMISSINGCELLS: 20
510      for (; rowIndex < 50; rowIndex++) {
511        Put put = new Put(Bytes.toBytes(rowIndex));
512        put.addColumn(family, column1, timestamps[0], value1);
513        put.addColumn(family, column2, timestamps[0], value2);
514        sourceTable.put(put);
515      }
516      // some rows only in the target table
517      // ROWSWITHDIFFS: 10
518      // SOURCEMISSINGROWS: 10
519      // SOURCEMISSINGCELLS: 20
520      for (; rowIndex < 60; rowIndex++) {
521        Put put = new Put(Bytes.toBytes(rowIndex));
522        put.addColumn(family, column1, timestamps[1], value1);
523        put.addColumn(family, column2, timestamps[1], value2);
524        targetTable.put(put);
525      }
526      // some rows with 1 missing cell in target table
527      // ROWSWITHDIFFS: 10
528      // TARGETMISSINGCELLS: 10
529      for (; rowIndex < 70; rowIndex++) {
530        Put sourcePut = new Put(Bytes.toBytes(rowIndex));
531        sourcePut.addColumn(family, column1, timestamps[0], value1);
532        sourcePut.addColumn(family, column2, timestamps[0], value2);
533        sourceTable.put(sourcePut);
534
535        Put targetPut = new Put(Bytes.toBytes(rowIndex));
536        targetPut.addColumn(family, column1, timestamps[1], value1);
537        targetTable.put(targetPut);
538      }
539      // some rows with 1 missing cell in source table
540      // ROWSWITHDIFFS: 10
541      // SOURCEMISSINGCELLS: 10
542      for (; rowIndex < 80; rowIndex++) {
543        Put sourcePut = new Put(Bytes.toBytes(rowIndex));
544        sourcePut.addColumn(family, column1, timestamps[0], value1);
545        sourceTable.put(sourcePut);
546
547        Put targetPut = new Put(Bytes.toBytes(rowIndex));
548        targetPut.addColumn(family, column1, timestamps[1], value1);
549        targetPut.addColumn(family, column2, timestamps[1], value2);
550        targetTable.put(targetPut);
551      }
552      // some rows differing only in timestamp
553      // ROWSWITHDIFFS: 10
554      // SOURCEMISSINGCELLS: 20
555      // TARGETMISSINGCELLS: 20
556      for (; rowIndex < 90; rowIndex++) {
557        Put sourcePut = new Put(Bytes.toBytes(rowIndex));
558        sourcePut.addColumn(family, column1, timestamps[0], column1);
559        sourcePut.addColumn(family, column2, timestamps[0], value2);
560        sourceTable.put(sourcePut);
561
562        Put targetPut = new Put(Bytes.toBytes(rowIndex));
563        targetPut.addColumn(family, column1, timestamps[1] + 1, column1);
564        targetPut.addColumn(family, column2, timestamps[1] - 1, value2);
565        targetTable.put(targetPut);
566      }
567      // some rows with different values
568      // ROWSWITHDIFFS: 10
569      // DIFFERENTCELLVALUES: 20
570      for (; rowIndex < numRows; rowIndex++) {
571        Put sourcePut = new Put(Bytes.toBytes(rowIndex));
572        sourcePut.addColumn(family, column1, timestamps[0], value1);
573        sourcePut.addColumn(family, column2, timestamps[0], value2);
574        sourceTable.put(sourcePut);
575
576        Put targetPut = new Put(Bytes.toBytes(rowIndex));
577        targetPut.addColumn(family, column1, timestamps[1], value3);
578        targetPut.addColumn(family, column2, timestamps[1], value3);
579        targetTable.put(targetPut);
580      }
581    }
582  }
583}