Merge branch 'apache:master' into master

pom.xml

@@ -559,7 +559,7 @@
     <hamcrest.version>2.2</hamcrest.version>
     <commons-cli.version>1.4</commons-cli.version>
     <netty.version>4.1.76.Final</netty.version>
-    <jetty.version>9.4.43.v20210629</jetty.version>
+    <jetty.version>9.4.49.v20220914</jetty.version>
     <jackson.version>2.13.2.1</jackson.version>
     <jline.version>2.14.6</jline.version>
     <snappy.version>1.1.7.7</snappy.version>

zookeeper-client/zookeeper-client-c/src/zookeeper.c

@@ -2751,11 +2751,6 @@ static int init_ssl_for_socket(zsock_t *fd, zhandle_t *zh, int fail_on_error) {
         OPENSSL_init_ssl(OPENSSL_INIT_LOAD_SSL_STRINGS | OPENSSL_INIT_LOAD_CRYPTO_STRINGS, NULL);
         method = TLS_client_method();
 #endif
-        if (FIPS_mode() == 0) {
-            LOG_INFO(LOGCALLBACK(zh), "FIPS mode is OFF ");
-        } else {
-            LOG_INFO(LOGCALLBACK(zh), "FIPS mode is ON ");
-        }
         fd->ssl_ctx = SSL_CTX_new(method);
         ctx = &fd->ssl_ctx;
 

zookeeper-client/zookeeper-client-c/tests/Util.cc

@@ -16,6 +16,8 @@
  * limitations under the License.
  */
 
+#include <time.h>
+
 #include "Util.h"
 #include "string.h"
 

zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md

@@ -1208,7 +1208,32 @@ property, when available, is noted below.
 
     The default value is false.
 
-
+* *serializeLastProcessedZxid.enabled*
+  (Jave system property: **zookeeper.serializeLastProcessedZxid.enabled**)
+  **New in 3.9.0:**
+  If enabled, ZooKeeper serializes the lastProcessedZxid when snapshot and deserializes it
+  when restore. Defaults to true. Needs to be enabled for performing snapshot and restore
+  via admin server commands, as there is no snapshot file name to extract the lastProcessedZxid.
+
+  This feature is backward and forward compatible. Here are the different scenarios.
+
+    1. Snapshot triggered by server internally
+       a. When loading old snapshot with new code, it will throw EOFException when trying to
+       read the non-exist lastProcessedZxid value, and the exception will be caught. 
+       The lastProcessedZxid will be set using the snapshot file name.
+
+       b. When loading new snapshot with old code, it will finish successfully after deserializing the 
+       digest value, the lastProcessedZxid at the end of snapshot file will be ignored.
+       The lastProcessedZxid will be set using the snapshot file name.
+
+    2. Sync up between leader and follower
+       The lastProcessedZxid will not be serialized by leader and deserialized by follower
+       in both new and old code. It will be set to the lastProcessedZxid sent from leader
+       via QuorumPacket.  
+
+   3. Snapshot triggered via admin server APIs
+      The feature flag need to be enabled for the snapshot command to work. 
+
 <a name="sc_clusterOptions"></a>
 
 #### Cluster Options
@@ -2087,6 +2112,20 @@ Both subsystems need to have sufficient amount of threads to achieve peak read t
 
 #### AdminServer configuration
 
+**New in 3.9.0:** The following
+options are used to configure the [AdminServer](#sc_adminserver).
+
+* *admin.snapshot.enabled* :
+  (Java system property: **zookeeper.admin.snapshot.enabled**)
+  The flag for enabling the snapshot command. Defaults to false. 
+  It will be enabled by default once the auth support for admin server commands 
+  is available.
+
+* *admin.snapshot.intervalInMS* :
+  (Java system property: **zookeeper.admin.snapshot.intervalInMS**)
+  The time interval for rate limiting snapshot command to protect the server.
+  Defaults to 5 mins.
+
 **New in 3.7.1:** The following
 options are used to configure the [AdminServer](#sc_adminserver).
 
@@ -2620,6 +2659,16 @@ Available commands include:
     Server information.
     Returns multiple fields giving a brief overview of server state.
 
+* *snapshot/snap* :
+  Takes a snapshot of the current server in the datadir and stream out data.
+  Optional query parameter:
+  "streaming": Boolean (defaults to true if the parameter is not present)
+  Returns the following via Http headers:
+  "last_zxid": String
+  "snapshot_size": String
+  Note: this API is rate-limited (once every 5 mins by default) to protect the server
+  from being over-loaded.
+
 * *stats/stat* :
     Same as *server_stats* but also returns the "connections" field (see *connections*
     for details).

zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java

@@ -1744,6 +1744,42 @@ public boolean deserializeZxidDigest(InputArchive ia, long startZxidOfSnapshot)
         }
     }
 
+    /**
+     * Serializes the lastProcessedZxid so we can get it from snapshot instead the snapshot file name.
+     * This is needed for performing snapshot and restore via admin server commands.
+     *
+     * @param oa the output stream to write to
+     * @return true if the lastProcessedZxid is serialized successfully, otherwise false
+     * @throws IOException if there is an I/O error
+     */
+    public boolean serializeLastProcessedZxid(final OutputArchive oa) throws IOException {
+        if (!ZooKeeperServer.isSerializeLastProcessedZxidEnabled()) {
+            return false;
+        }
+        oa.writeLong(lastProcessedZxid, "lastZxid");
+        return true;
+    }
+
+    /**
+     * Deserializes the lastProcessedZxid from the input stream and updates the lastProcessedZxid field.
+     *
+     * @param ia the input stream to read from
+     * @return true if lastProcessedZxid is deserialized successfully, otherwise false
+     * @throws IOException if there is an I/O error
+     */
+    public boolean deserializeLastProcessedZxid(final InputArchive ia)  throws IOException {
+        if (!ZooKeeperServer.isSerializeLastProcessedZxidEnabled()) {
+            return false;
+        }
+        try {
+            lastProcessedZxid = ia.readLong("lastZxid");
+        } catch (final EOFException e) {
+            LOG.warn("Got EOFException while reading the last processed zxid, likely due to reading an older snapshot.");
+            return false;
+        }
+        return true;
+    }
+
     /**
      * Compares the actual tree's digest with that in the snapshot.
      * Resets digestFromLoadedSnapshot after comparision.

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java

@@ -72,6 +72,8 @@ private ServerMetrics(MetricsProvider metricsProvider) {
         FSYNC_TIME = metricsContext.getSummary("fsynctime", DetailLevel.BASIC);
 
         SNAPSHOT_TIME = metricsContext.getSummary("snapshottime", DetailLevel.BASIC);
+        SNAPSHOT_ERROR_COUNT = metricsContext.getCounter("snapshot_error_count");
+        SNAPSHOT_RATE_LIMITED_COUNT = metricsContext.getCounter("snapshot_rate_limited_count");
         DB_INIT_TIME = metricsContext.getSummary("dbinittime", DetailLevel.BASIC);
         READ_LATENCY = metricsContext.getSummary("readlatency", DetailLevel.ADVANCED);
         UPDATE_LATENCY = metricsContext.getSummary("updatelatency", DetailLevel.ADVANCED);
@@ -276,6 +278,16 @@ private ServerMetrics(MetricsProvider metricsProvider) {
      */
     public final Summary SNAPSHOT_TIME;
 
+    /**
+     * Snapshot error count
+     */
+    public final Counter SNAPSHOT_ERROR_COUNT;
+
+    /**
+     * Snapshot rate limited count
+     */
+    public final Counter SNAPSHOT_RATE_LIMITED_COUNT;
+
     /**
      * Db init time (snapshot loading + txnlog replay)
      */

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -120,6 +120,9 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     public static final String ZOOKEEPER_DIGEST_ENABLED = "zookeeper.digest.enabled";
     private static boolean digestEnabled;
 
+    public static final String ZOOKEEPER_SERIALIZE_LAST_PROCESSED_ZXID_ENABLED = "zookeeper.serializeLastProcessedZxid.enabled";
+    private static boolean serializeLastProcessedZxidEnabled;
+
     // Add a enable/disable option for now, we should remove this one when
     // this feature is confirmed to be stable
     public static final String CLOSE_SESSION_TXN_ENABLED = "zookeeper.closeSessionTxn.enabled";
@@ -153,6 +156,9 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         closeSessionTxnEnabled = Boolean.parseBoolean(
                 System.getProperty(CLOSE_SESSION_TXN_ENABLED, "true"));
         LOG.info("{} = {}", CLOSE_SESSION_TXN_ENABLED, closeSessionTxnEnabled);
+
+        setSerializeLastProcessedZxidEnabled(Boolean.parseBoolean(
+                System.getProperty(ZOOKEEPER_SERIALIZE_LAST_PROCESSED_ZXID_ENABLED, "true")));
     }
 
     // @VisibleForTesting
@@ -535,23 +541,46 @@ public void loadData() throws IOException, InterruptedException {
         takeSnapshot();
     }
 
-    public void takeSnapshot() {
+    public void takeSnapshot() throws IOException {
         takeSnapshot(false);
     }
 
-    public void takeSnapshot(boolean syncSnap) {
+    public void takeSnapshot(boolean syncSnap) throws IOException {
+        takeSnapshot(syncSnap, true, false);
+    }
+
+    /**
+     * Takes a snapshot on the server.
+     *
+     * @param syncSnap syncSnap sync the snapshot immediately after write
+     * @param isSevere if true system exist, otherwise throw IOException
+     * @param fastForwardFromEdits whether fast forward database to the latest recorded transactions
+     *
+     * @return file snapshot file object
+     * @throws IOException
+     */
+    public synchronized File takeSnapshot(boolean syncSnap, boolean isSevere, boolean fastForwardFromEdits) throws IOException {
         long start = Time.currentElapsedTime();
+        File snapFile = null;
         try {
-            txnLogFactory.save(zkDb.getDataTree(), zkDb.getSessionWithTimeOuts(), syncSnap);
+            if (fastForwardFromEdits) {
+                zkDb.fastForwardDataBase();
+            }
+            snapFile = txnLogFactory.save(zkDb.getDataTree(), zkDb.getSessionWithTimeOuts(), syncSnap);
         } catch (IOException e) {
-            LOG.error("Severe unrecoverable error, exiting", e);
-            // This is a severe error that we cannot recover from,
-            // so we need to exit
-            ServiceUtils.requestSystemExit(ExitCode.TXNLOG_ERROR_TAKING_SNAPSHOT.getValue());
+            if (isSevere) {
+                LOG.error("Severe unrecoverable error, exiting", e);
+                // This is a severe error that we cannot recover from,
+                // so we need to exit
+                ServiceUtils.requestSystemExit(ExitCode.TXNLOG_ERROR_TAKING_SNAPSHOT.getValue());
+            } else {
+                throw e;
+            }
         }
         long elapsed = Time.currentElapsedTime() - start;
         LOG.info("Snapshot taken in {} ms", elapsed);
         ServerMetrics.getMetrics().SNAPSHOT_TIME.add(elapsed);
+        return snapFile;
     }
 
     public boolean shouldForceWriteInitialSnapshotAfterLeaderElection() {
@@ -2139,6 +2168,15 @@ public static void setDigestEnabled(boolean digestEnabled) {
         ZooKeeperServer.digestEnabled = digestEnabled;
     }
 
+    public static boolean isSerializeLastProcessedZxidEnabled() {
+        return serializeLastProcessedZxidEnabled;
+    }
+
+    public static void setSerializeLastProcessedZxidEnabled(boolean serializeLastZxidEnabled) {
+        serializeLastProcessedZxidEnabled = serializeLastZxidEnabled;
+        LOG.info("{} = {}", ZOOKEEPER_SERIALIZE_LAST_PROCESSED_ZXID_ENABLED, serializeLastZxidEnabled);
+    }
+
     /**
      * Trim a path to get the immediate predecessor.
      *

zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/CommandOutputter.java

@@ -18,6 +18,7 @@
 
 package org.apache.zookeeper.server.admin;
 
+import java.io.OutputStream;
 import java.io.PrintWriter;
 
 /**
@@ -31,6 +32,9 @@ public interface CommandOutputter {
     /** The MIME type of this output (e.g., "application/json") */
     String getContentType();
 
-    void output(CommandResponse response, PrintWriter pw);
+    /** Print out data as output */
+    default void output(CommandResponse response, PrintWriter pw) {}
 
+    /** Stream out data as output */
+    default void output(final CommandResponse response, final OutputStream os) {}
 }