Phoenix-7568 - Adding Replication Log Replay Implementation (#2278)

Phoenix-7568 Replication Log Replay Implementation

Author: Himanshu-g81

phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java

@@ -37,6 +37,7 @@
 import org.apache.phoenix.jdbc.ClusterRoleRecord;
 import org.apache.phoenix.jdbc.HAGroupStoreManager;
 import org.apache.phoenix.protobuf.ProtobufUtil;
+import org.apache.phoenix.replication.reader.ReplicationLogReplayService;
 import org.apache.phoenix.util.ClientUtil;
 import org.apache.phoenix.util.SchemaUtil;
 import org.apache.phoenix.util.ServerUtil;
@@ -62,10 +63,14 @@ public void start(CoprocessorEnvironment env) throws IOException {
         this.metricsSource = MetricsPhoenixCoprocessorSourceFactory
                                 .getInstance().getMetadataCachingSource();
         this.zkUrl = getLocalZkUrl(conf);
+        // Start replication log replay
+        ReplicationLogReplayService.getInstance(conf).start();
     }
 
     @Override
     public void stop(CoprocessorEnvironment env) throws IOException {
+        // Stop replication log replay
+        ReplicationLogReplayService.getInstance(conf).stop();
         RegionServerCoprocessor.super.stop(env);
         ServerUtil.ConnectionFactory.shutdown();
     }

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java

@@ -58,9 +58,9 @@ public class ReplicationLogGroup {
         "phoenix.replication.log.standby.hdfs.url";
     public static final String REPLICATION_FALLBACK_HDFS_URL_KEY =
         "phoenix.replication.log.fallback.hdfs.url";
-    public static final String REPLICATION_NUM_SHARDS_KEY = "phoenix.replication.log.shards";
-    public static final int DEFAULT_REPLICATION_NUM_SHARDS = 1000;
-    public static final int MAX_REPLICATION_NUM_SHARDS = 100000;
+    public static final String REPLICATION_LOG_ROTATION_TIME_MS_KEY =
+        "phoenix.replication.log.rotation.time.ms";
+    public static final long DEFAULT_REPLICATION_LOG_ROTATION_TIME_MS = 60 * 1000L;
     public static final String REPLICATION_LOG_ROTATION_SIZE_BYTES_KEY =
         "phoenix.replication.log.rotation.size.bytes";
     public static final long DEFAULT_REPLICATION_LOG_ROTATION_SIZE_BYTES = 256 * 1024 * 1024L;
@@ -86,8 +86,7 @@ public class ReplicationLogGroup {
         "phoenix.replication.log.retry.delay.ms";
     public static final long DEFAULT_REPLICATION_LOG_RETRY_DELAY_MS = 100L;
 
-    public static final String SHARD_DIR_FORMAT = "%05d";
-    public static final String FILE_NAME_FORMAT = "%d-%s.plog";
+    public static final String FILE_NAME_FORMAT = "%d_%s.plog";
 
     /** Cache of ReplicationLogGroup instances by HA Group ID */
     protected static final ConcurrentHashMap<String, ReplicationLogGroup> INSTANCES =

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java

@@ -127,6 +127,7 @@ public abstract class ReplicationLogGroupWriter {
     protected Disruptor<LogEvent> disruptor;
     protected RingBuffer<LogEvent> ringBuffer;
     protected volatile boolean closed = false;
+    protected ReplicationShardDirectoryManager replicationShardDirectoryManager;
 
     /** The reason for requesting a log rotation. */
     protected enum RotationReason {
@@ -179,6 +180,7 @@ protected ReplicationLogGroupWriter(ReplicationLogGroup logGroup) {
     /** Initialize the writer. */
     public void init() throws IOException {
         initializeFileSystems();
+        initializeReplicationShardDirectoryManager();
         // Start time based rotation.
         lastRotationTime.set(EnvironmentEdgeManager.currentTimeMillis());
         startRotationExecutor();
@@ -251,6 +253,12 @@ public void sync() throws IOException {
     /** Initialize file systems needed by this writer implementation. */
     protected abstract void initializeFileSystems() throws IOException;
 
+    /**
+     * Initialize the {@link ReplicationShardDirectoryManager} to manage file to shard directory
+     * mapping
+     */
+    protected abstract void initializeReplicationShardDirectoryManager();
+
     /**
      * Create a new log writer for rotation.
      */