fix: disabling incompatible mutation metrics for HBase 1.3 or lower (#3599)

Due to lack of custom counter support in HBase.

#3596

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/java-bigtable-hbase/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://github.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: vermas2012

hbase-migration-tools/bigtable-hbase-replication/README.md

@@ -61,7 +61,10 @@ data. Near zero downtime migrations include the following steps:
    and service account json file.
 4. Add a CBT replication peer in HBase. On HBase shell
    execute `add_peer '2', ENDPOINT_CLASSNAME => 'com.google.cloud.bigtable.hbase2_x.replication.HbaseToCloudBigtableReplicationEndpoint'`
-   . Please use endpoint class `com.google.cloud.bigtable.hbase1_x.replication.HbaseToCloudBigtableReplicationEndpoint` for HBase 1.x clusters. Use add_peer options to enable replication for select tables.
+   . Please use endpoint
+   class `com.google.cloud.bigtable.hbase1_x.replication.HbaseToCloudBigtableReplicationEndpoint`
+   for HBase 1.x clusters. Use add_peer options to enable replication for select
+   tables.
 5. Immediately disable the CBT replication peer, this allows WAL logs to
    accumulate on HDFS. On HBase shell execute:  `disable_peer '2'`
 6. Check the replicated tables by executing `list_replicated_tables` and enable
@@ -111,9 +114,11 @@ classpath.
 </property>
 ```
 
-We recommend specifying a single-cluster routing [application profile](https://cloud.google.com/bigtable/docs/app-profiles#routing) by setting config key 
+We recommend specifying a single-cluster
+routing [application profile](https://cloud.google.com/bigtable/docs/app-profiles#routing)
+by setting config key
 `google.bigtable.app_profile.id`. A single-cluster routing application profile
-preserves order of mutations between HBase and Cloud Bigtable. 
+preserves order of mutations between HBase and Cloud Bigtable.
 
 Next, you should configure Cloud Bigtable authentication. Create a service
 account and download a json file as shown
@@ -236,7 +241,9 @@ in `hbase-site.xml` but we recommend setting it during peer creation.
 Enabling/disabling dry run mode during peer creation can avoid restarting the
 HBase cluster to pickup changes to `hbase-site.xml` file. Enable dry run mode by
 running the following command to add Cloud Bigtable replication peer (please
-change the endpoint class to `com.google.cloud.bigtable.hbase1_x.replication.HbaseToCloudBigtableReplicationEndpoint` for HBase 1.x):
+change the endpoint class
+to `com.google.cloud.bigtable.hbase1_x.replication.HbaseToCloudBigtableReplicationEndpoint`
+for HBase 1.x):
 
 ```
 add_peer 'peer_id',
@@ -283,7 +290,8 @@ are 3 kinds of metrics that the replication library will publish:
 2. Cloud Bigtable client side metrics. These will include latencies and failures
    of various CBT APIs.
 3. Custom metrics from the replication library. For example,
-   NumberOfIncompatibleMutations.
+   NumberOfIncompatibleMutations. Please note that cusotm metrics support is
+   available for HBase 1.4 or newer.
 
 Please refer to javadocs for class HBaseToCloudBigtableReplicationMetrics for
 list of available metrics.

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/main/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpoint.java

@@ -37,12 +37,12 @@ public class HbaseToCloudBigtableReplicationEndpoint extends BaseReplicationEndp
 
   public HbaseToCloudBigtableReplicationEndpoint() {
     cloudBigtableReplicator = new CloudBigtableReplicator();
-    metricsExporter = new HBaseMetricsExporter();
+    metricsExporter = HBaseMetricsExporter.create();
   }
 
   @Override
   protected synchronized void doStart() {
-    metricsExporter.setMetricsSource(ctx.getMetrics());
+    metricsExporter.init(ctx);
     cloudBigtableReplicator.start(ctx.getConfiguration(), metricsExporter);
     notifyStarted();
   }

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/main/java/com/google/cloud/bigtable/hbase1_x/replication/metrics/HBaseMetricsExporter.java

@@ -17,21 +17,57 @@
 package com.google.cloud.bigtable.hbase1_x.replication.metrics;
 
 import com.google.bigtable.repackaged.com.google.api.core.InternalApi;
+import com.google.bigtable.repackaged.com.google.api.core.InternalExtensionOnly;
 import com.google.cloud.bigtable.hbase.replication.metrics.MetricsExporter;
+import java.lang.reflect.Method;
+import org.apache.hadoop.hbase.replication.ReplicationEndpoint.Context;
 import org.apache.hadoop.hbase.replication.regionserver.MetricsSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** HBaseMetricsExporter implements MetricExporter which bridges with MetricsSource. */
 @InternalApi
 public class HBaseMetricsExporter implements MetricsExporter {
+
+  private static final Logger LOG = LoggerFactory.getLogger(HBaseMetricsExporter.class);
+
+  // Force the use of static factory method to create instances.
+  @InternalExtensionOnly
+  protected HBaseMetricsExporter() {}
+
   // same pattern as used by HbaseInterClusterRepl
   private MetricsSource metricsSource;
 
-  public void setMetricsSource(MetricsSource metricsSource) {
-    this.metricsSource = metricsSource;
+  public void init(Context ctx) {
+    this.metricsSource = ctx.getMetrics();
   }
 
   @Override
   public void incCounters(String counterName, long delta) {
     metricsSource.incCounters(counterName, delta);
   }
+
+  /**
+   * Creates the right implementation for HBaseMetricsExporter. The incCounter method used to
+   * integrate with HBase metrics was introduced in HBase 1.4, so when running on HBase version 1.3
+   * or lower, we need to skip incrementing the counters. More details on:
+   * https://github.com/googleapis/java-bigtable-hbase/issues/3596
+   */
+  public static HBaseMetricsExporter create() {
+    // TODO: Define configuration that allows users to inject a custom implementation of
+    //  HBaseMetricsExporter
+    try {
+      Method method = MetricsSource.class.getMethod("incCounters", String.class, long.class);
+      // HBase version > 1.4, supports incCounters, return normal MetricsExporter.
+      return new HBaseMetricsExporter();
+    } catch (NoSuchMethodException e) {
+      // HBase version <1.4 : HBase does not support generic counters. Revert to no-op metrics
+      // exporter.
+      LOG.warn(
+          "Can not find MetricsSource.incCounters method, probably running HBase 1.3 or older."
+              + " Disabling metrics for IncompatibleMutations. Please refer to "
+              + "https://github.com/googleapis/java-bigtable-hbase/issues/3596 for details.");
+      return new NoOpHBaseMetricsExporter();
+    }
+  }
 }

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/main/java/com/google/cloud/bigtable/hbase1_x/replication/metrics/NoOpHBaseMetricsExporter.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.bigtable.hbase1_x.replication.metrics;
+
+import com.google.bigtable.repackaged.com.google.api.core.InternalApi;
+
+/**
+ * No-op implemnentation of MetricsExporter interface. To be used where incCounters method from
+ * HBase MetricsSource is not available.
+ */
+@InternalApi
+public class NoOpHBaseMetricsExporter extends HBaseMetricsExporter {
+
+  // Use the HBaseMetricsExporter.create method to create instances of this class.
+  @InternalApi
+  NoOpHBaseMetricsExporter() {}
+
+  @Override
+  public void incCounters(String counterName, long delta) {}
+}

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/metrics/HBaseMetricsExporterTest.java

@@ -16,10 +16,15 @@
 
 package com.google.cloud.bigtable.hbase1_x.replication.metrics;
 
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
+import org.apache.hadoop.hbase.replication.ReplicationEndpoint.Context;
 import org.apache.hadoop.hbase.replication.regionserver.MetricsSource;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -33,6 +38,7 @@
 public class HBaseMetricsExporterTest {
   @Rule public final MockitoRule mockitoRule = MockitoJUnit.rule();
 
+  @Mock Context context;
   @Mock MetricsSource metricsSource;
 
   HBaseMetricsExporter hbaseMetricsExporter;
@@ -42,8 +48,22 @@ public class HBaseMetricsExporterTest {
 
   @Before
   public void setUp() {
-    hbaseMetricsExporter = new HBaseMetricsExporter();
-    hbaseMetricsExporter.setMetricsSource(metricsSource);
+    when(context.getMetrics()).thenReturn(metricsSource);
+    hbaseMetricsExporter = HBaseMetricsExporter.create();
+    hbaseMetricsExporter.init(context);
+  }
+
+  @After
+  public void tearDown() {
+    reset(context, metricsSource);
+  }
+
+  @Test
+  public void testCreate() {
+    // Make sure that create returns an HBaseMetricsExporter object. There is no good way to test
+    // the other case where incCounter method is not available as it requires adding hbase <1.4 to
+    // the classpath along with hbase 1.4.
+    assertEquals(HBaseMetricsExporter.class, hbaseMetricsExporter.getClass());
   }
 
   @Test

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/metrics/HBaseToCloudBigtableReplicationMetrics.java

@@ -16,6 +16,11 @@
 
 package com.google.cloud.bigtable.hbase.replication.metrics;
 
+/**
+ * Metrics exported by the Cloud Bigtable replication endpoint. Please note that custom replication
+ * metrics are only supported by HBase 1.4 and newer. Please see
+ * https://github.com/googleapis/java-bigtable-hbase/issues/3596 for more details.
+ */
 public class HBaseToCloudBigtableReplicationMetrics {
 
   // Static class for listing all the metrics