diff --git a/CHANGELOG.md b/CHANGELOG.md
index 96e1511eead..e0e64087c04 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,15 +1,20 @@
 # Changelog
 
 ## [Unreleased]
+- Add configuration of Consolidation Request Contract Address via genesis configuration [#7647](https://github.com/hyperledger/besu/pull/7647)
+
 
 ### Upcoming Breaking Changes
 - k8s (KUBERNETES) Nat method is now deprecated and will be removed in a future release
 
 ### Breaking Changes
+- Besu will now fail to start if any plugins encounter errors during initialization. To allow Besu to continue running despite plugin errors, use the `--plugin-continue-on-error` option. [#7662](https://github.com/hyperledger/besu/pull/7662)
 
 ### Additions and Improvements
 - Remove privacy test classes support [#7569](https://github.com/hyperledger/besu/pull/7569)
 - Add Blob Transaction Metrics [#7622](https://github.com/hyperledger/besu/pull/7622)
+- Implemented support for emptyBlockPeriodSeconds in QBFT [#6965](https://github.com/hyperledger/besu/pull/6965)
+
 
 ### Bug fixes
 - Fix mounted data path directory permissions for besu user [#7575](https://github.com/hyperledger/besu/pull/7575)
diff --git a/acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/node/ThreadBesuNodeRunner.java b/acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/node/ThreadBesuNodeRunner.java
index 5d78f1460c4..90de0b0e952 100644
--- a/acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/node/ThreadBesuNodeRunner.java
+++ b/acceptance-tests/dsl/src/main/java/org/hyperledger/besu/tests/acceptance/dsl/node/ThreadBesuNodeRunner.java
@@ -503,8 +503,9 @@ public BesuPluginContextImpl providePluginContext(
       besuPluginContext.addService(PermissioningService.class, permissioningService);
       besuPluginContext.addService(PrivacyPluginService.class, new PrivacyPluginServiceImpl());
 
-      besuPluginContext.registerPlugins(
+      besuPluginContext.initialize(
           new PluginConfiguration.Builder().pluginsDir(pluginsPath).build());
+      besuPluginContext.registerPlugins();
       commandLine.parseArgs(extraCLIOptions.toArray(new String[0]));
 
       // register built-in plugins
diff --git a/acceptance-tests/test-plugins/src/main/java/org/hyperledger/besu/tests/acceptance/plugins/TestPicoCLIPlugin.java b/acceptance-tests/test-plugins/src/main/java/org/hyperledger/besu/tests/acceptance/plugins/TestPicoCLIPlugin.java
index 0db5281537c..375fbd490ec 100644
--- a/acceptance-tests/test-plugins/src/main/java/org/hyperledger/besu/tests/acceptance/plugins/TestPicoCLIPlugin.java
+++ b/acceptance-tests/test-plugins/src/main/java/org/hyperledger/besu/tests/acceptance/plugins/TestPicoCLIPlugin.java
@@ -1,5 +1,5 @@
 /*
- * Copyright ConsenSys AG.
+ * Copyright contributors to Hyperledger Besu.
  *
  * 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
@@ -32,16 +32,25 @@
 public class TestPicoCLIPlugin implements BesuPlugin {
   private static final Logger LOG = LoggerFactory.getLogger(TestPicoCLIPlugin.class);
 
+  private static final String UNSET = "UNSET";
+  private static final String FAIL_REGISTER = "FAILREGISTER";
+  private static final String FAIL_BEFORE_EXTERNAL_SERVICES = "FAILBEFOREEXTERNALSERVICES";
+  private static final String FAIL_START = "FAILSTART";
+  private static final String FAIL_AFTER_EXTERNAL_SERVICE_POST_MAIN_LOOP =
+      "FAILAFTEREXTERNALSERVICEPOSTMAINLOOP";
+  private static final String FAIL_STOP = "FAILSTOP";
+  private static final String PLUGIN_LIFECYCLE_PREFIX = "pluginLifecycle.";
+
   @Option(
       names = {"--Xplugin-test-option"},
       hidden = true,
-      defaultValue = "UNSET")
+      defaultValue = UNSET)
   String testOption = System.getProperty("testPicoCLIPlugin.testOption");
 
   @Option(
       names = {"--plugin-test-stable-option"},
       hidden = true,
-      defaultValue = "UNSET")
+      defaultValue = UNSET)
   String stableOption = "";
 
   private String state = "uninited";
@@ -52,7 +61,7 @@ public void register(final BesuContext context) {
     LOG.info("Registering.  Test Option is '{}'", testOption);
     state = "registering";
 
-    if ("FAILREGISTER".equals(testOption)) {
+    if (FAIL_REGISTER.equals(testOption)) {
       state = "failregister";
       throw new RuntimeException("I was told to fail at registration");
     }
@@ -66,12 +75,26 @@ public void register(final BesuContext context) {
     state = "registered";
   }
 
+  @Override
+  public void beforeExternalServices() {
+    LOG.info("Before external services. Test Option is '{}'", testOption);
+    state = "beforeExternalServices";
+
+    if (FAIL_BEFORE_EXTERNAL_SERVICES.equals(testOption)) {
+      state = "failbeforeExternalServices";
+      throw new RuntimeException("I was told to fail before external services");
+    }
+
+    writeSignal("beforeExternalServices");
+    state = "beforeExternalServicesFinished";
+  }
+
   @Override
   public void start() {
     LOG.info("Starting.  Test Option is '{}'", testOption);
     state = "starting";
 
-    if ("FAILSTART".equals(testOption)) {
+    if (FAIL_START.equals(testOption)) {
       state = "failstart";
       throw new RuntimeException("I was told to fail at startup");
     }
@@ -80,12 +103,26 @@ public void start() {
     state = "started";
   }
 
+  @Override
+  public void afterExternalServicePostMainLoop() {
+    LOG.info("After external services post main loop. Test Option is '{}'", testOption);
+    state = "afterExternalServicePostMainLoop";
+
+    if (FAIL_AFTER_EXTERNAL_SERVICE_POST_MAIN_LOOP.equals(testOption)) {
+      state = "failafterExternalServicePostMainLoop";
+      throw new RuntimeException("I was told to fail after external services post main loop");
+    }
+
+    writeSignal("afterExternalServicePostMainLoop");
+    state = "afterExternalServicePostMainLoopFinished";
+  }
+
   @Override
   public void stop() {
     LOG.info("Stopping.  Test Option is '{}'", testOption);
     state = "stopping";
 
-    if ("FAILSTOP".equals(testOption)) {
+    if (FAIL_STOP.equals(testOption)) {
       state = "failstop";
       throw new RuntimeException("I was told to fail at stop");
     }
@@ -103,7 +140,7 @@ public String getState() {
   @SuppressWarnings("ResultOfMethodCallIgnored")
   private void writeSignal(final String signal) {
     try {
-      final File callbackFile = new File(callbackDir, "pluginLifecycle." + signal);
+      final File callbackFile = new File(callbackDir, PLUGIN_LIFECYCLE_PREFIX + signal);
       if (!callbackFile.getParentFile().exists()) {
         callbackFile.getParentFile().mkdirs();
         callbackFile.getParentFile().deleteOnExit();
diff --git a/acceptance-tests/test-plugins/src/test/java/org/hyperledger/besu/services/BesuPluginContextImplTest.java b/acceptance-tests/test-plugins/src/test/java/org/hyperledger/besu/services/BesuPluginContextImplTest.java
index 7e277041776..a6d167665ff 100644
--- a/acceptance-tests/test-plugins/src/test/java/org/hyperledger/besu/services/BesuPluginContextImplTest.java
+++ b/acceptance-tests/test-plugins/src/test/java/org/hyperledger/besu/services/BesuPluginContextImplTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright ConsenSys AG.
+ * Copyright contributors to Hyperledger Besu.
  *
  * 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
@@ -40,8 +40,31 @@
 
 public class BesuPluginContextImplTest {
   private static final Path DEFAULT_PLUGIN_DIRECTORY = Paths.get(".");
+  private static final String TEST_PICO_CLI_PLUGIN = "TestPicoCLIPlugin";
+  private static final String TEST_PICO_CLI_PLUGIN_TEST_OPTION = "testPicoCLIPlugin.testOption";
+  private static final String FAIL_REGISTER = "FAILREGISTER";
+  private static final String FAIL_START = "FAILSTART";
+  private static final String FAIL_STOP = "FAILSTOP";
+  private static final String FAIL_BEFORE_EXTERNAL_SERVICES = "FAILBEFOREEXTERNALSERVICES";
+  private static final String FAIL_BEFORE_MAIN_LOOP = "FAILBEFOREMAINLOOP";
+  private static final String FAIL_AFTER_EXTERNAL_SERVICE_POST_MAIN_LOOP =
+      "FAILAFTEREXTERNALSERVICEPOSTMAINLOOP";
+  private static final String NON_EXISTENT_PLUGIN = "NonExistentPlugin";
+  private static final String REGISTERED = "registered";
+  private static final String STARTED = "started";
+  private static final String STOPPED = "stopped";
+  private static final String FAIL_START_STATE = "failstart";
+  private static final String FAIL_STOP_STATE = "failstop";
+
   private BesuPluginContextImpl contextImpl;
 
+  private static final PluginConfiguration DEFAULT_CONFIGURATION =
+      PluginConfiguration.builder()
+          .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+          .externalPluginsEnabled(true)
+          .continueOnPluginError(true)
+          .build();
+
   @BeforeAll
   public static void createFakePluginDir() throws IOException {
     if (System.getProperty("besu.plugins.dir") == null) {
@@ -53,7 +76,7 @@ public static void createFakePluginDir() throws IOException {
 
   @AfterEach
   public void clearTestPluginState() {
-    System.clearProperty("testPicoCLIPlugin.testOption");
+    System.clearProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION);
   }
 
   @BeforeEach
@@ -64,31 +87,31 @@ void setup() {
   @Test
   public void verifyEverythingGoesSmoothly() {
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
-    contextImpl.registerPlugins(
-        PluginConfiguration.builder().pluginsDir(DEFAULT_PLUGIN_DIRECTORY).build());
+    contextImpl.initialize(DEFAULT_CONFIGURATION);
+    contextImpl.registerPlugins();
     assertThat(contextImpl.getRegisteredPlugins()).isNotEmpty();
 
     final Optional<TestPicoCLIPlugin> testPluginOptional =
         findTestPlugin(contextImpl.getRegisteredPlugins(), TestPicoCLIPlugin.class);
     assertThat(testPluginOptional).isPresent();
     final TestPicoCLIPlugin testPicoCLIPlugin = testPluginOptional.get();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("registered");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(REGISTERED);
 
     contextImpl.beforeExternalServices();
     contextImpl.startPlugins();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("started");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(STARTED);
 
     contextImpl.stopPlugins();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("stopped");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(STOPPED);
   }
 
   @Test
   public void registrationErrorsHandledSmoothly() {
-    System.setProperty("testPicoCLIPlugin.testOption", "FAILREGISTER");
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_REGISTER);
 
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
-    contextImpl.registerPlugins(
-        PluginConfiguration.builder().pluginsDir(DEFAULT_PLUGIN_DIRECTORY).build());
+    contextImpl.initialize(DEFAULT_CONFIGURATION);
+    contextImpl.registerPlugins();
     assertThat(contextImpl.getRegisteredPlugins()).isNotInstanceOfAny(TestPicoCLIPlugin.class);
 
     contextImpl.beforeExternalServices();
@@ -103,11 +126,11 @@ public void registrationErrorsHandledSmoothly() {
 
   @Test
   public void startErrorsHandledSmoothly() {
-    System.setProperty("testPicoCLIPlugin.testOption", "FAILSTART");
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_START);
 
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
-    contextImpl.registerPlugins(
-        PluginConfiguration.builder().pluginsDir(DEFAULT_PLUGIN_DIRECTORY).build());
+    contextImpl.initialize(DEFAULT_CONFIGURATION);
+    contextImpl.registerPlugins();
     assertThat(contextImpl.getRegisteredPlugins())
         .extracting("class")
         .contains(TestPicoCLIPlugin.class);
@@ -116,11 +139,11 @@ public void startErrorsHandledSmoothly() {
         findTestPlugin(contextImpl.getRegisteredPlugins(), TestPicoCLIPlugin.class);
     assertThat(testPluginOptional).isPresent();
     final TestPicoCLIPlugin testPicoCLIPlugin = testPluginOptional.get();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("registered");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(REGISTERED);
 
     contextImpl.beforeExternalServices();
     contextImpl.startPlugins();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("failstart");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(FAIL_START_STATE);
     assertThat(contextImpl.getRegisteredPlugins()).isNotInstanceOfAny(TestPicoCLIPlugin.class);
 
     contextImpl.stopPlugins();
@@ -129,11 +152,11 @@ public void startErrorsHandledSmoothly() {
 
   @Test
   public void stopErrorsHandledSmoothly() {
-    System.setProperty("testPicoCLIPlugin.testOption", "FAILSTOP");
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_STOP);
 
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
-    contextImpl.registerPlugins(
-        PluginConfiguration.builder().pluginsDir(DEFAULT_PLUGIN_DIRECTORY).build());
+    contextImpl.initialize(DEFAULT_CONFIGURATION);
+    contextImpl.registerPlugins();
     assertThat(contextImpl.getRegisteredPlugins())
         .extracting("class")
         .contains(TestPicoCLIPlugin.class);
@@ -142,22 +165,20 @@ public void stopErrorsHandledSmoothly() {
         findTestPlugin(contextImpl.getRegisteredPlugins(), TestPicoCLIPlugin.class);
     assertThat(testPluginOptional).isPresent();
     final TestPicoCLIPlugin testPicoCLIPlugin = testPluginOptional.get();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("registered");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(REGISTERED);
 
     contextImpl.beforeExternalServices();
     contextImpl.startPlugins();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("started");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(STARTED);
 
     contextImpl.stopPlugins();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("failstop");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(FAIL_STOP_STATE);
   }
 
   @Test
   public void lifecycleExceptions() throws Throwable {
-    final ThrowableAssert.ThrowingCallable registerPlugins =
-        () ->
-            contextImpl.registerPlugins(
-                PluginConfiguration.builder().pluginsDir(DEFAULT_PLUGIN_DIRECTORY).build());
+    contextImpl.initialize(DEFAULT_CONFIGURATION);
+    final ThrowableAssert.ThrowingCallable registerPlugins = () -> contextImpl.registerPlugins();
 
     assertThatExceptionOfType(IllegalStateException.class).isThrownBy(contextImpl::startPlugins);
     assertThatExceptionOfType(IllegalStateException.class).isThrownBy(contextImpl::stopPlugins);
@@ -179,30 +200,27 @@ public void lifecycleExceptions() throws Throwable {
 
   @Test
   public void shouldRegisterAllPluginsWhenNoPluginsOption() {
-    final PluginConfiguration config =
-        PluginConfiguration.builder().pluginsDir(DEFAULT_PLUGIN_DIRECTORY).build();
-
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
-    contextImpl.registerPlugins(config);
+    contextImpl.initialize(DEFAULT_CONFIGURATION);
+    contextImpl.registerPlugins();
     final Optional<TestPicoCLIPlugin> testPluginOptional =
         findTestPlugin(contextImpl.getRegisteredPlugins(), TestPicoCLIPlugin.class);
     assertThat(testPluginOptional).isPresent();
     final TestPicoCLIPlugin testPicoCLIPlugin = testPluginOptional.get();
-    assertThat(testPicoCLIPlugin.getState()).isEqualTo("registered");
+    assertThat(testPicoCLIPlugin.getState()).isEqualTo(REGISTERED);
   }
 
   @Test
   public void shouldRegisterOnlySpecifiedPluginWhenPluginsOptionIsSet() {
-    final PluginConfiguration config = createConfigurationForSpecificPlugin("TestPicoCLIPlugin");
-
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
-    contextImpl.registerPlugins(config);
+    contextImpl.initialize(createConfigurationForSpecificPlugin(TEST_PICO_CLI_PLUGIN));
+    contextImpl.registerPlugins();
 
     final Optional<TestPicoCLIPlugin> requestedPlugin =
         findTestPlugin(contextImpl.getRegisteredPlugins(), TestPicoCLIPlugin.class);
 
     assertThat(requestedPlugin).isPresent();
-    assertThat(requestedPlugin.get().getState()).isEqualTo("registered");
+    assertThat(requestedPlugin.get().getState()).isEqualTo(REGISTERED);
 
     final Optional<TestPicoCLIPlugin> nonRequestedPlugin =
         findTestPlugin(contextImpl.getRegisteredPlugins(), TestBesuEventsPlugin.class);
@@ -212,9 +230,9 @@ public void shouldRegisterOnlySpecifiedPluginWhenPluginsOptionIsSet() {
 
   @Test
   public void shouldNotRegisterUnspecifiedPluginsWhenPluginsOptionIsSet() {
-    final PluginConfiguration config = createConfigurationForSpecificPlugin("TestPicoCLIPlugin");
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
-    contextImpl.registerPlugins(config);
+    contextImpl.initialize(createConfigurationForSpecificPlugin(TEST_PICO_CLI_PLUGIN));
+    contextImpl.registerPlugins();
 
     final Optional<TestPicoCLIPlugin> nonRequestedPlugin =
         findTestPlugin(contextImpl.getRegisteredPlugins(), TestBesuEventsPlugin.class);
@@ -223,13 +241,12 @@ public void shouldNotRegisterUnspecifiedPluginsWhenPluginsOptionIsSet() {
 
   @Test
   void shouldThrowExceptionIfExplicitlySpecifiedPluginNotFound() {
-    PluginConfiguration config = createConfigurationForSpecificPlugin("NonExistentPlugin");
-
+    contextImpl.initialize(createConfigurationForSpecificPlugin(NON_EXISTENT_PLUGIN));
     String exceptionMessage =
-        assertThrows(NoSuchElementException.class, () -> contextImpl.registerPlugins(config))
+        assertThrows(NoSuchElementException.class, () -> contextImpl.registerPlugins())
             .getMessage();
     final String expectedMessage =
-        "The following requested plugins were not found: NonExistentPlugin";
+        "The following requested plugins were not found: " + NON_EXISTENT_PLUGIN;
     assertThat(exceptionMessage).isEqualTo(expectedMessage);
     assertThat(contextImpl.getRegisteredPlugins()).isEmpty();
   }
@@ -241,19 +258,180 @@ void shouldNotRegisterAnyPluginsIfExternalPluginsDisabled() {
             .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
             .externalPluginsEnabled(false)
             .build();
-    contextImpl.registerPlugins(config);
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
     assertThat(contextImpl.getRegisteredPlugins().isEmpty()).isTrue();
   }
 
   @Test
   void shouldRegisterPluginsIfExternalPluginsEnabled() {
+    contextImpl.initialize(DEFAULT_CONFIGURATION);
+    contextImpl.registerPlugins();
+    assertThat(contextImpl.getRegisteredPlugins().isEmpty()).isFalse();
+  }
+
+  @Test
+  void shouldHaltOnRegisterErrorWhenFlagIsFalse() {
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_REGISTER);
+
     PluginConfiguration config =
         PluginConfiguration.builder()
+            .requestedPlugins(List.of(new PluginInfo(TEST_PICO_CLI_PLUGIN)))
             .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
-            .externalPluginsEnabled(true)
+            .continueOnPluginError(false)
             .build();
-    contextImpl.registerPlugins(config);
-    assertThat(contextImpl.getRegisteredPlugins().isEmpty()).isFalse();
+
+    contextImpl.initialize(config);
+
+    String errorMessage =
+        String.format("Error registering plugin of type %s", TestPicoCLIPlugin.class.getName());
+    assertThatExceptionOfType(RuntimeException.class)
+        .isThrownBy(() -> contextImpl.registerPlugins())
+        .withMessageContaining(errorMessage);
+  }
+
+  @Test
+  void shouldNotHaltOnRegisterErrorWhenFlagIsTrue() {
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_REGISTER);
+
+    PluginConfiguration config =
+        PluginConfiguration.builder()
+            .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+            .continueOnPluginError(true)
+            .build();
+
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
+    assertThat(contextImpl.getRegisteredPlugins()).isNotInstanceOfAny(TestPicoCLIPlugin.class);
+  }
+
+  @Test
+  void shouldHaltOnBeforeExternalServicesErrorWhenFlagIsFalse() {
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_BEFORE_EXTERNAL_SERVICES);
+
+    PluginConfiguration config =
+        PluginConfiguration.builder()
+            .requestedPlugins(List.of(new PluginInfo(TEST_PICO_CLI_PLUGIN)))
+            .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+            .continueOnPluginError(false)
+            .build();
+
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
+
+    String errorMessage =
+        String.format(
+            "Error calling `beforeExternalServices` on plugin of type %s",
+            TestPicoCLIPlugin.class.getName());
+    assertThatExceptionOfType(RuntimeException.class)
+        .isThrownBy(() -> contextImpl.beforeExternalServices())
+        .withMessageContaining(errorMessage);
+  }
+
+  @Test
+  void shouldNotHaltOnBeforeExternalServicesErrorWhenFlagIsTrue() {
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_BEFORE_EXTERNAL_SERVICES);
+
+    PluginConfiguration config =
+        PluginConfiguration.builder()
+            .requestedPlugins(List.of(new PluginInfo(TEST_PICO_CLI_PLUGIN)))
+            .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+            .continueOnPluginError(true)
+            .build();
+
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
+    contextImpl.beforeExternalServices();
+
+    assertThat(contextImpl.getRegisteredPlugins()).isNotInstanceOfAny(TestPicoCLIPlugin.class);
+  }
+
+  @Test
+  void shouldHaltOnBeforeMainLoopErrorWhenFlagIsFalse() {
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_START);
+
+    PluginConfiguration config =
+        PluginConfiguration.builder()
+            .requestedPlugins(List.of(new PluginInfo(TEST_PICO_CLI_PLUGIN)))
+            .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+            .continueOnPluginError(false)
+            .build();
+
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
+    contextImpl.beforeExternalServices();
+
+    String errorMessage =
+        String.format("Error starting plugin of type %s", TestPicoCLIPlugin.class.getName());
+    assertThatExceptionOfType(RuntimeException.class)
+        .isThrownBy(() -> contextImpl.startPlugins())
+        .withMessageContaining(errorMessage);
+  }
+
+  @Test
+  void shouldNotHaltOnBeforeMainLoopErrorWhenFlagIsTrue() {
+    System.setProperty(TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_BEFORE_MAIN_LOOP);
+
+    PluginConfiguration config =
+        PluginConfiguration.builder()
+            .requestedPlugins(List.of(new PluginInfo(TEST_PICO_CLI_PLUGIN)))
+            .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+            .continueOnPluginError(true)
+            .build();
+
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
+    contextImpl.beforeExternalServices();
+    contextImpl.startPlugins();
+
+    assertThat(contextImpl.getRegisteredPlugins()).isNotInstanceOfAny(TestPicoCLIPlugin.class);
+  }
+
+  @Test
+  void shouldHaltOnAfterExternalServicePostMainLoopErrorWhenFlagIsFalse() {
+    System.setProperty(
+        TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_AFTER_EXTERNAL_SERVICE_POST_MAIN_LOOP);
+
+    PluginConfiguration config =
+        PluginConfiguration.builder()
+            .requestedPlugins(List.of(new PluginInfo(TEST_PICO_CLI_PLUGIN)))
+            .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+            .continueOnPluginError(false)
+            .build();
+
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
+    contextImpl.beforeExternalServices();
+    contextImpl.startPlugins();
+
+    String errorMessage =
+        String.format(
+            "Error calling `afterExternalServicePostMainLoop` on plugin of type %s",
+            TestPicoCLIPlugin.class.getName());
+    assertThatExceptionOfType(RuntimeException.class)
+        .isThrownBy(() -> contextImpl.afterExternalServicesMainLoop())
+        .withMessageContaining(errorMessage);
+  }
+
+  @Test
+  void shouldNotHaltOnAfterExternalServicePostMainLoopErrorWhenFlagIsTrue() {
+    System.setProperty(
+        TEST_PICO_CLI_PLUGIN_TEST_OPTION, FAIL_AFTER_EXTERNAL_SERVICE_POST_MAIN_LOOP);
+
+    PluginConfiguration config =
+        PluginConfiguration.builder()
+            .requestedPlugins(List.of(new PluginInfo(TEST_PICO_CLI_PLUGIN)))
+            .pluginsDir(DEFAULT_PLUGIN_DIRECTORY)
+            .continueOnPluginError(true)
+            .build();
+
+    contextImpl.initialize(config);
+    contextImpl.registerPlugins();
+    contextImpl.beforeExternalServices();
+    contextImpl.startPlugins();
+    contextImpl.afterExternalServicesMainLoop();
+
+    assertThat(contextImpl.getRegisteredPlugins()).isNotInstanceOfAny(TestPicoCLIPlugin.class);
   }
 
   private PluginConfiguration createConfigurationForSpecificPlugin(final String pluginName) {
diff --git a/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java b/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java
index 79879538bd5..e93518cad43 100644
--- a/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java
+++ b/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java
@@ -118,7 +118,6 @@
 import org.hyperledger.besu.ethereum.core.MiningParametersMetrics;
 import org.hyperledger.besu.ethereum.core.PrivacyParameters;
 import org.hyperledger.besu.ethereum.core.VersionMetadata;
-import org.hyperledger.besu.ethereum.core.plugins.PluginConfiguration;
 import org.hyperledger.besu.ethereum.eth.sync.SyncMode;
 import org.hyperledger.besu.ethereum.eth.sync.SynchronizerConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
@@ -1080,9 +1079,8 @@ private IExecutionStrategy createExecuteTask(final IExecutionStrategy nextStep)
 
   private IExecutionStrategy createPluginRegistrationTask(final IExecutionStrategy nextStep) {
     return parseResult -> {
-      PluginConfiguration configuration =
-          PluginsConfigurationOptions.fromCommandLine(parseResult.commandSpec().commandLine());
-      besuPluginContext.registerPlugins(configuration);
+      besuPluginContext.initialize(PluginsConfigurationOptions.fromCommandLine(commandLine));
+      besuPluginContext.registerPlugins();
       commandLine.setExecutionStrategy(nextStep);
       return commandLine.execute(parseResult.originalArgs().toArray(new String[0]));
     };
diff --git a/besu/src/main/java/org/hyperledger/besu/cli/DefaultCommandValues.java b/besu/src/main/java/org/hyperledger/besu/cli/DefaultCommandValues.java
index ba05f455246..8f83c71a787 100644
--- a/besu/src/main/java/org/hyperledger/besu/cli/DefaultCommandValues.java
+++ b/besu/src/main/java/org/hyperledger/besu/cli/DefaultCommandValues.java
@@ -128,6 +128,9 @@ public interface DefaultCommandValues {
   /** The constant DEFAULT_PLUGINS_OPTION_NAME. */
   String DEFAULT_PLUGINS_OPTION_NAME = "--plugins";
 
+  /** The constant DEFAULT_CONTINUE_ON_PLUGIN_ERROR_OPTION_NAME. */
+  String DEFAULT_CONTINUE_ON_PLUGIN_ERROR_OPTION_NAME = "--plugin-continue-on-error";
+
   /** The constant DEFAULT_PLUGINS_EXTERNAL_ENABLED_OPTION_NAME. */
   String DEFAULT_PLUGINS_EXTERNAL_ENABLED_OPTION_NAME = "--Xplugins-external-enabled";
 
diff --git a/besu/src/main/java/org/hyperledger/besu/cli/options/stable/PluginsConfigurationOptions.java b/besu/src/main/java/org/hyperledger/besu/cli/options/stable/PluginsConfigurationOptions.java
index 47df831c577..25893fff895 100644
--- a/besu/src/main/java/org/hyperledger/besu/cli/options/stable/PluginsConfigurationOptions.java
+++ b/besu/src/main/java/org/hyperledger/besu/cli/options/stable/PluginsConfigurationOptions.java
@@ -14,6 +14,7 @@
  */
 package org.hyperledger.besu.cli.options.stable;
 
+import static org.hyperledger.besu.cli.DefaultCommandValues.DEFAULT_CONTINUE_ON_PLUGIN_ERROR_OPTION_NAME;
 import static org.hyperledger.besu.cli.DefaultCommandValues.DEFAULT_PLUGINS_EXTERNAL_ENABLED_OPTION_NAME;
 import static org.hyperledger.besu.cli.DefaultCommandValues.DEFAULT_PLUGINS_OPTION_NAME;
 
@@ -27,7 +28,7 @@
 
 import picocli.CommandLine;
 
-/** The Plugins Options options. */
+/** The Plugins options. */
 public class PluginsConfigurationOptions implements CLIOptions<PluginConfiguration> {
 
   @CommandLine.Option(
@@ -44,9 +45,17 @@ public class PluginsConfigurationOptions implements CLIOptions<PluginConfigurati
       split = ",",
       hidden = true,
       converter = PluginInfoConverter.class,
-      arity = "1..*")
+      arity = "1")
   private List<PluginInfo> plugins;
 
+  @CommandLine.Option(
+      names = {DEFAULT_CONTINUE_ON_PLUGIN_ERROR_OPTION_NAME},
+      description =
+          "Allow Besu startup even if any plugins fail to initialize correctly (default: ${DEFAULT-VALUE})",
+      defaultValue = "false",
+      arity = "1")
+  private final Boolean continueOnPluginError = false;
+
   /** Default Constructor. */
   public PluginsConfigurationOptions() {}
 
@@ -55,6 +64,7 @@ public PluginConfiguration toDomainObject() {
     return new PluginConfiguration.Builder()
         .externalPluginsEnabled(externalPluginsEnabled)
         .requestedPlugins(plugins)
+        .continueOnPluginError(continueOnPluginError)
         .build();
   }
 
@@ -66,10 +76,15 @@ public PluginConfiguration toDomainObject() {
   public void validate(final CommandLine commandLine) {
     String errorMessage =
         String.format(
-            "%s option can only be used when %s is true",
-            DEFAULT_PLUGINS_OPTION_NAME, DEFAULT_PLUGINS_EXTERNAL_ENABLED_OPTION_NAME);
+            "%s and %s option can only be used when %s is true",
+            DEFAULT_PLUGINS_OPTION_NAME,
+            DEFAULT_CONTINUE_ON_PLUGIN_ERROR_OPTION_NAME,
+            DEFAULT_PLUGINS_EXTERNAL_ENABLED_OPTION_NAME);
     CommandLineUtils.failIfOptionDoesntMeetRequirement(
-        commandLine, errorMessage, externalPluginsEnabled, List.of(DEFAULT_PLUGINS_OPTION_NAME));
+        commandLine,
+        errorMessage,
+        externalPluginsEnabled,
+        List.of(DEFAULT_PLUGINS_OPTION_NAME, DEFAULT_CONTINUE_ON_PLUGIN_ERROR_OPTION_NAME));
   }
 
   @Override
@@ -92,9 +107,14 @@ public static PluginConfiguration fromCommandLine(final CommandLine commandLine)
         CommandLineUtils.getOptionValueOrDefault(
             commandLine, DEFAULT_PLUGINS_EXTERNAL_ENABLED_OPTION_NAME, Boolean::parseBoolean);
 
+    boolean continueOnPluginError =
+        CommandLineUtils.getOptionValueOrDefault(
+            commandLine, DEFAULT_CONTINUE_ON_PLUGIN_ERROR_OPTION_NAME, Boolean::parseBoolean);
+
     return new PluginConfiguration.Builder()
         .requestedPlugins(plugins)
         .externalPluginsEnabled(externalPluginsEnabled)
+        .continueOnPluginError(continueOnPluginError)
         .build();
   }
 }
diff --git a/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java b/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java
index 95bbe0a2b1f..816d9df00a1 100644
--- a/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java
+++ b/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/SynchronizerOptions.java
@@ -314,6 +314,13 @@ public void parseBlockPropagationRange(final String arg) {
       description = "Snap sync enabled for BFT chains (default: ${DEFAULT-VALUE})")
   private Boolean snapsyncBftEnabled = SnapSyncConfiguration.DEFAULT_SNAP_SYNC_BFT_ENABLED;
 
+  @CommandLine.Option(
+      names = {"--Xpeertask-system-enabled"},
+      hidden = true,
+      description =
+          "Temporary feature toggle to enable using the new peertask system (default: ${DEFAULT-VALUE})")
+  private final Boolean isPeerTaskSystemEnabled = false;
+
   private SynchronizerOptions() {}
 
   /**
@@ -334,6 +341,15 @@ public boolean isSnapSyncBftEnabled() {
     return snapsyncBftEnabled;
   }
 
+  /**
+   * Flag to indicate whether the peer task system should be used where available
+   *
+   * @return true if the peer task system should be used where available
+   */
+  public boolean isPeerTaskSystemEnabled() {
+    return isPeerTaskSystemEnabled;
+  }
+
   /**
    * Create synchronizer options.
    *
@@ -420,7 +436,7 @@ public SynchronizerConfiguration.Builder toDomainObject() {
             .isSnapSyncBftEnabled(snapsyncBftEnabled)
             .build());
     builder.checkpointPostMergeEnabled(checkpointPostMergeSyncEnabled);
-
+    builder.isPeerTaskSystemEnabled(isPeerTaskSystemEnabled);
     return builder;
   }
 
diff --git a/besu/src/main/java/org/hyperledger/besu/controller/QbftBesuControllerBuilder.java b/besu/src/main/java/org/hyperledger/besu/controller/QbftBesuControllerBuilder.java
index 3d3412d4860..60eac179965 100644
--- a/besu/src/main/java/org/hyperledger/besu/controller/QbftBesuControllerBuilder.java
+++ b/besu/src/main/java/org/hyperledger/besu/controller/QbftBesuControllerBuilder.java
@@ -288,12 +288,18 @@ protected MiningCoordinator createMiningCoordinator(
     protocolContext
         .getBlockchain()
         .observeBlockAdded(
-            o ->
-                miningParameters.setBlockPeriodSeconds(
-                    qbftForksSchedule
-                        .getFork(o.getBlock().getHeader().getNumber() + 1)
-                        .getValue()
-                        .getBlockPeriodSeconds()));
+            o -> {
+              miningParameters.setBlockPeriodSeconds(
+                  qbftForksSchedule
+                      .getFork(o.getBlock().getHeader().getNumber() + 1)
+                      .getValue()
+                      .getBlockPeriodSeconds());
+              miningParameters.setEmptyBlockPeriodSeconds(
+                  qbftForksSchedule
+                      .getFork(o.getBlock().getHeader().getNumber() + 1)
+                      .getValue()
+                      .getEmptyBlockPeriodSeconds());
+            });
 
     if (syncState.isInitialSyncPhaseDone()) {
       miningCoordinator.enable();
@@ -422,8 +428,9 @@ private static MinedBlockObserver blockLogger(
     return block ->
         LOG.info(
             String.format(
-                "%s #%,d / %d tx / %d pending / %,d (%01.1f%%) gas / (%s)",
+                "%s %s #%,d / %d tx / %d pending / %,d (%01.1f%%) gas / (%s)",
                 block.getHeader().getCoinbase().equals(localAddress) ? "Produced" : "Imported",
+                block.getBody().getTransactions().size() == 0 ? "empty block" : "block",
                 block.getHeader().getNumber(),
                 block.getBody().getTransactions().size(),
                 transactionPool.count(),
diff --git a/besu/src/main/java/org/hyperledger/besu/services/BesuPluginContextImpl.java b/besu/src/main/java/org/hyperledger/besu/services/BesuPluginContextImpl.java
index c89733f9354..ad4dbd44819 100644
--- a/besu/src/main/java/org/hyperledger/besu/services/BesuPluginContextImpl.java
+++ b/besu/src/main/java/org/hyperledger/besu/services/BesuPluginContextImpl.java
@@ -56,6 +56,8 @@ public class BesuPluginContextImpl implements BesuContext, PluginVersionsProvide
   private enum Lifecycle {
     /** Uninitialized lifecycle. */
     UNINITIALIZED,
+    /** Initialized lifecycle. */
+    INITIALIZED,
     /** Registering lifecycle. */
     REGISTERING,
     /** Registered lifecycle. */
@@ -83,6 +85,7 @@ private enum Lifecycle {
   private final List<BesuPlugin> registeredPlugins = new ArrayList<>();
 
   private final List<String> pluginVersions = new ArrayList<>();
+  private PluginConfiguration config;
 
   /** Instantiates a new Besu plugin context. */
   public BesuPluginContextImpl() {}
@@ -116,19 +119,30 @@ private List<BesuPlugin> detectPlugins(final PluginConfiguration config) {
     return StreamSupport.stream(serviceLoader.spliterator(), false).toList();
   }
 
+  /**
+   * Initializes the plugin context with the provided {@link PluginConfiguration}.
+   *
+   * @param config the plugin configuration
+   * @throws IllegalStateException if the system is not in the UNINITIALIZED state.
+   */
+  public void initialize(final PluginConfiguration config) {
+    checkState(
+        state == Lifecycle.UNINITIALIZED,
+        "Besu plugins have already been initialized. Cannot register additional plugins.");
+    this.config = config;
+    state = Lifecycle.INITIALIZED;
+  }
+
   /**
    * Registers plugins based on the provided {@link PluginConfiguration}. This method finds plugins
    * according to the configuration settings, filters them if necessary and then registers the
    * filtered or found plugins
    *
-   * @param config The configuration settings used to find and filter plugins for registration. The
-   *     configuration includes the plugin directory and any configured plugin identifiers if
-   *     applicable.
    * @throws IllegalStateException if the system is not in the UNINITIALIZED state.
    */
-  public void registerPlugins(final PluginConfiguration config) {
+  public void registerPlugins() {
     checkState(
-        state == Lifecycle.UNINITIALIZED,
+        state == Lifecycle.INITIALIZED,
         "Besu plugins have already been registered. Cannot register additional plugins.");
     state = Lifecycle.REGISTERING;
 
@@ -192,14 +206,18 @@ private void registerPlugins(final List<BesuPlugin> pluginsToRegister) {
   private boolean registerPlugin(final BesuPlugin plugin) {
     try {
       plugin.register(this);
-      LOG.info("Registered plugin of type {}.", plugin.getClass().getName());
       pluginVersions.add(plugin.getVersion());
+      LOG.info("Registered plugin of type {}.", plugin.getClass().getName());
     } catch (final Exception e) {
-      LOG.error(
-          "Error registering plugin of type "
-              + plugin.getClass().getName()
-              + ", start and stop will not be called.",
-          e);
+      if (config.isContinueOnPluginError()) {
+        LOG.error(
+            "Error registering plugin of type {}, start and stop will not be called.",
+            plugin.getClass().getName(),
+            e);
+      } else {
+        throw new RuntimeException(
+            "Error registering plugin of type " + plugin.getClass().getName(), e);
+      }
       return false;
     }
     return true;
@@ -223,15 +241,20 @@ public void beforeExternalServices() {
         LOG.debug(
             "beforeExternalServices called on plugin of type {}.", plugin.getClass().getName());
       } catch (final Exception e) {
-        LOG.error(
-            "Error calling `beforeExternalServices` on plugin of type "
-                + plugin.getClass().getName()
-                + ", stop will not be called.",
-            e);
-        pluginsIterator.remove();
+        if (config.isContinueOnPluginError()) {
+          LOG.error(
+              "Error calling `beforeExternalServices` on plugin of type {}, start will not be called.",
+              plugin.getClass().getName(),
+              e);
+          pluginsIterator.remove();
+        } else {
+          throw new RuntimeException(
+              "Error calling `beforeExternalServices` on plugin of type "
+                  + plugin.getClass().getName(),
+              e);
+        }
       }
     }
-
     LOG.debug("Plugin startup complete.");
     state = Lifecycle.BEFORE_EXTERNAL_SERVICES_FINISHED;
   }
@@ -253,12 +276,16 @@ public void startPlugins() {
         plugin.start();
         LOG.debug("Started plugin of type {}.", plugin.getClass().getName());
       } catch (final Exception e) {
-        LOG.error(
-            "Error starting plugin of type "
-                + plugin.getClass().getName()
-                + ", stop will not be called.",
-            e);
-        pluginsIterator.remove();
+        if (config.isContinueOnPluginError()) {
+          LOG.error(
+              "Error starting plugin of type {}, stop will not be called.",
+              plugin.getClass().getName(),
+              e);
+          pluginsIterator.remove();
+        } else {
+          throw new RuntimeException(
+              "Error starting plugin of type " + plugin.getClass().getName(), e);
+        }
       }
     }
 
@@ -279,8 +306,20 @@ public void afterExternalServicesMainLoop() {
       final BesuPlugin plugin = pluginsIterator.next();
       try {
         plugin.afterExternalServicePostMainLoop();
-      } finally {
-        pluginsIterator.remove();
+      } catch (final Exception e) {
+        if (config.isContinueOnPluginError()) {
+          LOG.error(
+              "Error calling `afterExternalServicePostMainLoop` on plugin of type "
+                  + plugin.getClass().getName()
+                  + ", stop will not be called.",
+              e);
+          pluginsIterator.remove();
+        } else {
+          throw new RuntimeException(
+              "Error calling `afterExternalServicePostMainLoop` on plugin of type "
+                  + plugin.getClass().getName(),
+              e);
+        }
       }
     }
   }
diff --git a/besu/src/test/java/org/hyperledger/besu/cli/CommandTestAbstract.java b/besu/src/test/java/org/hyperledger/besu/cli/CommandTestAbstract.java
index 5b1274389d6..65e51c92139 100644
--- a/besu/src/test/java/org/hyperledger/besu/cli/CommandTestAbstract.java
+++ b/besu/src/test/java/org/hyperledger/besu/cli/CommandTestAbstract.java
@@ -138,15 +138,18 @@ public abstract class CommandTestAbstract {
   private static final Logger TEST_LOGGER = LoggerFactory.getLogger(CommandTestAbstract.class);
 
   protected static final int POA_BLOCK_PERIOD_SECONDS = 5;
+  protected static final int POA_EMPTY_BLOCK_PERIOD_SECONDS = 50;
   protected static final JsonObject VALID_GENESIS_QBFT_POST_LONDON =
       (new JsonObject())
           .put(
               "config",
               new JsonObject()
                   .put("londonBlock", 0)
+                  .put("qbft", new JsonObject().put("blockperiodseconds", POA_BLOCK_PERIOD_SECONDS))
                   .put(
                       "qbft",
-                      new JsonObject().put("blockperiodseconds", POA_BLOCK_PERIOD_SECONDS)));
+                      new JsonObject()
+                          .put("xemptyblockperiodseconds", POA_EMPTY_BLOCK_PERIOD_SECONDS)));
   protected static final JsonObject VALID_GENESIS_IBFT2_POST_LONDON =
       (new JsonObject())
           .put(
diff --git a/besu/src/test/java/org/hyperledger/besu/cli/PluginsOptionsTest.java b/besu/src/test/java/org/hyperledger/besu/cli/PluginsOptionsTest.java
index 1adfb585e9d..566c7621281 100644
--- a/besu/src/test/java/org/hyperledger/besu/cli/PluginsOptionsTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/cli/PluginsOptionsTest.java
@@ -33,7 +33,7 @@ public class PluginsOptionsTest extends CommandTestAbstract {
   @Test
   public void shouldParsePluginOptionForSinglePlugin() {
     parseCommand("--plugins", "pluginA");
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
     assertThat(pluginConfigurationArgumentCaptor.getValue().getRequestedPlugins())
         .isEqualTo(List.of("pluginA"));
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
@@ -43,7 +43,7 @@ public void shouldParsePluginOptionForSinglePlugin() {
   @Test
   public void shouldParsePluginOptionForMultiplePlugins() {
     parseCommand("--plugins", "pluginA,pluginB");
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
     assertThat(pluginConfigurationArgumentCaptor.getValue().getRequestedPlugins())
         .isEqualTo(List.of("pluginA", "pluginB"));
 
@@ -54,7 +54,7 @@ public void shouldParsePluginOptionForMultiplePlugins() {
   @Test
   public void shouldNotUsePluginOptionWhenNoPluginsSpecified() {
     parseCommand();
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
     assertThat(pluginConfigurationArgumentCaptor.getValue().getRequestedPlugins())
         .isEqualTo(List.of());
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
@@ -64,7 +64,7 @@ public void shouldNotUsePluginOptionWhenNoPluginsSpecified() {
   @Test
   public void shouldNotParseAnyPluginsWhenPluginOptionIsEmpty() {
     parseCommand("--plugins", "");
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
     assertThat(pluginConfigurationArgumentCaptor.getValue().getRequestedPlugins())
         .isEqualTo(List.of());
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
@@ -74,7 +74,7 @@ public void shouldNotParseAnyPluginsWhenPluginOptionIsEmpty() {
   @Test
   public void shouldParsePluginsExternalEnabledOptionWhenFalse() {
     parseCommand("--Xplugins-external-enabled=false");
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
 
     assertThat(pluginConfigurationArgumentCaptor.getValue().isExternalPluginsEnabled())
         .isEqualTo(false);
@@ -86,7 +86,7 @@ public void shouldParsePluginsExternalEnabledOptionWhenFalse() {
   @Test
   public void shouldParsePluginsExternalEnabledOptionWhenTrue() {
     parseCommand("--Xplugins-external-enabled=true");
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
 
     assertThat(pluginConfigurationArgumentCaptor.getValue().isExternalPluginsEnabled())
         .isEqualTo(true);
@@ -98,7 +98,7 @@ public void shouldParsePluginsExternalEnabledOptionWhenTrue() {
   @Test
   public void shouldEnablePluginsExternalByDefault() {
     parseCommand();
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
     assertThat(pluginConfigurationArgumentCaptor.getValue().isExternalPluginsEnabled())
         .isEqualTo(true);
 
@@ -109,10 +109,43 @@ public void shouldEnablePluginsExternalByDefault() {
   @Test
   public void shouldFailWhenPluginsIsDisabledAndPluginsExplicitlyRequested() {
     parseCommand("--Xplugins-external-enabled=false", "--plugins", "pluginA");
-    verify(mockBesuPluginContext).registerPlugins(pluginConfigurationArgumentCaptor.capture());
 
     assertThat(commandOutput.toString(UTF_8)).isEmpty();
     assertThat(commandErrorOutput.toString(UTF_8))
-        .contains("--plugins option can only be used when --Xplugins-external-enabled is true");
+        .contains(
+            "--plugins and --plugin-continue-on-error option can only be used when --Xplugins-external-enabled is true");
+  }
+
+  @Test
+  public void shouldHaveContinueOnErrorFalseByDefault() {
+    parseCommand();
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
+    assertThat(pluginConfigurationArgumentCaptor.getValue().isContinueOnPluginError())
+        .isEqualTo(false);
+
+    assertThat(commandOutput.toString(UTF_8)).isEmpty();
+    assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();
+  }
+
+  @Test
+  public void shouldUseContinueOnErrorWhenTrue() {
+    parseCommand("--plugin-continue-on-error=true");
+    verify(mockBesuPluginContext).initialize(pluginConfigurationArgumentCaptor.capture());
+
+    assertThat(pluginConfigurationArgumentCaptor.getValue().isContinueOnPluginError())
+        .isEqualTo(true);
+
+    assertThat(commandOutput.toString(UTF_8)).isEmpty();
+    assertThat(commandErrorOutput.toString(UTF_8)).isEmpty();
+  }
+
+  @Test
+  public void shouldFailWhenPluginsIsDisabledAnHaltOnErrorTrue() {
+    parseCommand("--Xplugins-external-enabled=false", "--plugin-continue-on-error=true");
+
+    assertThat(commandOutput.toString(UTF_8)).isEmpty();
+    assertThat(commandErrorOutput.toString(UTF_8))
+        .contains(
+            "--plugins and --plugin-continue-on-error option can only be used when --Xplugins-external-enabled is true");
   }
 }
diff --git a/besu/src/test/resources/everything_config.toml b/besu/src/test/resources/everything_config.toml
index e3d7a3f28d3..e89442f7f78 100644
--- a/besu/src/test/resources/everything_config.toml
+++ b/besu/src/test/resources/everything_config.toml
@@ -226,6 +226,7 @@ Xsecp256k1-native-enabled=false
 Xaltbn128-native-enabled=false
 Xsnapsync-server-enabled=true
 Xbonsai-full-flat-db-enabled=true
+Xpeertask-system-enabled=false
 
 # compatibility flags
 compatibility-eth64-forkid-enabled=false
@@ -246,4 +247,5 @@ Xevm-jumpdest-cache-weight-kb=32000
 
 # plugins
 Xplugins-external-enabled=true
-plugins=["none"]
\ No newline at end of file
+plugins=["none"]
+plugin-continue-on-error=false
\ No newline at end of file
diff --git a/config/src/main/java/org/hyperledger/besu/config/BftConfigOptions.java b/config/src/main/java/org/hyperledger/besu/config/BftConfigOptions.java
index 58df7be8490..8ad35bf2d33 100644
--- a/config/src/main/java/org/hyperledger/besu/config/BftConfigOptions.java
+++ b/config/src/main/java/org/hyperledger/besu/config/BftConfigOptions.java
@@ -37,6 +37,13 @@ public interface BftConfigOptions {
    */
   int getBlockPeriodSeconds();
 
+  /**
+   * Gets empty block period seconds.
+   *
+   * @return the empty block period seconds
+   */
+  int getEmptyBlockPeriodSeconds();
+
   /**
    * Gets block period milliseconds. For TESTING only. If set then blockperiodseconds is ignored.
    *
diff --git a/config/src/main/java/org/hyperledger/besu/config/BftFork.java b/config/src/main/java/org/hyperledger/besu/config/BftFork.java
index fdaa8f2fa9d..9083bf339ec 100644
--- a/config/src/main/java/org/hyperledger/besu/config/BftFork.java
+++ b/config/src/main/java/org/hyperledger/besu/config/BftFork.java
@@ -41,6 +41,9 @@ public class BftFork implements Fork {
   /** The constant BLOCK_PERIOD_SECONDS_KEY. */
   public static final String BLOCK_PERIOD_SECONDS_KEY = "blockperiodseconds";
 
+  /** The constant EMPTY_BLOCK_PERIOD_SECONDS_KEY. */
+  public static final String EMPTY_BLOCK_PERIOD_SECONDS_KEY = "xemptyblockperiodseconds";
+
   /** The constant BLOCK_PERIOD_MILLISECONDS_KEY. */
   public static final String BLOCK_PERIOD_MILLISECONDS_KEY = "xblockperiodmilliseconds";
 
@@ -86,6 +89,16 @@ public OptionalInt getBlockPeriodSeconds() {
     return JsonUtil.getPositiveInt(forkConfigRoot, BLOCK_PERIOD_SECONDS_KEY);
   }
 
+  /**
+   * Gets empty block period seconds.
+   *
+   * @return the empty block period seconds
+   */
+  public OptionalInt getEmptyBlockPeriodSeconds() {
+    // It can be 0 to disable custom empty block periods
+    return JsonUtil.getInt(forkConfigRoot, EMPTY_BLOCK_PERIOD_SECONDS_KEY);
+  }
+
   /**
    * Gets block period milliseconds. Experimental for test scenarios only.
    *
diff --git a/config/src/main/java/org/hyperledger/besu/config/GenesisConfigOptions.java b/config/src/main/java/org/hyperledger/besu/config/GenesisConfigOptions.java
index 5fa670fbee8..2301a785e55 100644
--- a/config/src/main/java/org/hyperledger/besu/config/GenesisConfigOptions.java
+++ b/config/src/main/java/org/hyperledger/besu/config/GenesisConfigOptions.java
@@ -553,4 +553,11 @@ default boolean isConsensusMigration() {
    * @return the deposit address
    */
   Optional<Address> getDepositContractAddress();
+
+  /**
+   * The consolidation request contract address
+   *
+   * @return the consolidation request contract address
+   */
+  Optional<Address> getConsolidationRequestContractAddress();
 }
diff --git a/config/src/main/java/org/hyperledger/besu/config/JsonBftConfigOptions.java b/config/src/main/java/org/hyperledger/besu/config/JsonBftConfigOptions.java
index b1c8630b399..fb9c8d6c796 100644
--- a/config/src/main/java/org/hyperledger/besu/config/JsonBftConfigOptions.java
+++ b/config/src/main/java/org/hyperledger/besu/config/JsonBftConfigOptions.java
@@ -34,6 +34,8 @@ public class JsonBftConfigOptions implements BftConfigOptions {
 
   private static final long DEFAULT_EPOCH_LENGTH = 30_000;
   private static final int DEFAULT_BLOCK_PERIOD_SECONDS = 1;
+  // 0 keeps working as before, increase to activate it
+  private static final int DEFAULT_EMPTY_BLOCK_PERIOD_SECONDS = 0;
   private static final int DEFAULT_BLOCK_PERIOD_MILLISECONDS = 0; // Experimental for test only
   private static final int DEFAULT_ROUND_EXPIRY_SECONDS = 1;
   // In a healthy network this can be very small. This default limit will allow for suitable
@@ -67,6 +69,12 @@ public int getBlockPeriodSeconds() {
         bftConfigRoot, "blockperiodseconds", DEFAULT_BLOCK_PERIOD_SECONDS);
   }
 
+  @Override
+  public int getEmptyBlockPeriodSeconds() {
+    return JsonUtil.getInt(
+        bftConfigRoot, "xemptyblockperiodseconds", DEFAULT_EMPTY_BLOCK_PERIOD_SECONDS);
+  }
+
   @Override
   public long getBlockPeriodMilliseconds() {
     return JsonUtil.getLong(
diff --git a/config/src/main/java/org/hyperledger/besu/config/JsonGenesisConfigOptions.java b/config/src/main/java/org/hyperledger/besu/config/JsonGenesisConfigOptions.java
index 248056fec2f..2cf598da854 100644
--- a/config/src/main/java/org/hyperledger/besu/config/JsonGenesisConfigOptions.java
+++ b/config/src/main/java/org/hyperledger/besu/config/JsonGenesisConfigOptions.java
@@ -52,6 +52,8 @@ public class JsonGenesisConfigOptions implements GenesisConfigOptions {
   private static final String WITHDRAWAL_REQUEST_CONTRACT_ADDRESS_KEY =
       "withdrawalrequestcontractaddress";
   private static final String DEPOSIT_CONTRACT_ADDRESS_KEY = "depositcontractaddress";
+  private static final String CONSOLIDATION_REQUEST_CONTRACT_ADDRESS_KEY =
+      "consolidationrequestcontractaddress";
 
   private final ObjectNode configRoot;
   private final Map<String, String> configOverrides = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
@@ -463,6 +465,13 @@ public Optional<Address> getDepositContractAddress() {
     return inputAddress.map(Address::fromHexString);
   }
 
+  @Override
+  public Optional<Address> getConsolidationRequestContractAddress() {
+    Optional<String> inputAddress =
+        JsonUtil.getString(configRoot, CONSOLIDATION_REQUEST_CONTRACT_ADDRESS_KEY);
+    return inputAddress.map(Address::fromHexString);
+  }
+
   @Override
   public Map<String, Object> asMap() {
     final ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
@@ -517,6 +526,8 @@ public Map<String, Object> asMap() {
     getWithdrawalRequestContractAddress()
         .ifPresent(l -> builder.put("withdrawalRequestContractAddress", l));
     getDepositContractAddress().ifPresent(l -> builder.put("depositContractAddress", l));
+    getConsolidationRequestContractAddress()
+        .ifPresent(l -> builder.put("consolidationRequestContractAddress", l));
 
     if (isClique()) {
       builder.put("clique", getCliqueConfigOptions().asMap());
diff --git a/config/src/main/java/org/hyperledger/besu/config/StubGenesisConfigOptions.java b/config/src/main/java/org/hyperledger/besu/config/StubGenesisConfigOptions.java
index f8c2a52d8cb..9948c54e226 100644
--- a/config/src/main/java/org/hyperledger/besu/config/StubGenesisConfigOptions.java
+++ b/config/src/main/java/org/hyperledger/besu/config/StubGenesisConfigOptions.java
@@ -483,6 +483,11 @@ public Optional<Address> getDepositContractAddress() {
     return Optional.empty();
   }
 
+  @Override
+  public Optional<Address> getConsolidationRequestContractAddress() {
+    return Optional.empty();
+  }
+
   /**
    * Homestead block stub genesis config options.
    *
diff --git a/config/src/test/java/org/hyperledger/besu/config/GenesisConfigOptionsTest.java b/config/src/test/java/org/hyperledger/besu/config/GenesisConfigOptionsTest.java
index 219ea4fcf8a..69494a9c9c4 100644
--- a/config/src/test/java/org/hyperledger/besu/config/GenesisConfigOptionsTest.java
+++ b/config/src/test/java/org/hyperledger/besu/config/GenesisConfigOptionsTest.java
@@ -382,6 +382,33 @@ void asMapIncludesDepositContractAddress() {
         .containsValue(Address.ZERO);
   }
 
+  @Test
+  void shouldGetConsolidationRequestContractAddress() {
+    final GenesisConfigOptions config =
+        fromConfigOptions(
+            singletonMap(
+                "consolidationRequestContractAddress",
+                "0x00000000219ab540356cbb839cbe05303d7705fa"));
+    assertThat(config.getConsolidationRequestContractAddress())
+        .hasValue(Address.fromHexString("0x00000000219ab540356cbb839cbe05303d7705fa"));
+  }
+
+  @Test
+  void shouldNotHaveConsolidationRequestContractAddressWhenEmpty() {
+    final GenesisConfigOptions config = fromConfigOptions(emptyMap());
+    assertThat(config.getConsolidationRequestContractAddress()).isEmpty();
+  }
+
+  @Test
+  void asMapIncludesConsolidationRequestContractAddress() {
+    final GenesisConfigOptions config =
+        fromConfigOptions(Map.of("consolidationRequestContractAddress", "0x0"));
+
+    assertThat(config.asMap())
+        .containsOnlyKeys("consolidationRequestContractAddress")
+        .containsValue(Address.ZERO);
+  }
+
   private GenesisConfigOptions fromConfigOptions(final Map<String, Object> configOptions) {
     final ObjectNode rootNode = JsonUtil.createEmptyObjectNode();
     final ObjectNode options = JsonUtil.objectNodeFromMap(configOptions);
diff --git a/config/src/test/java/org/hyperledger/besu/config/JsonBftConfigOptionsTest.java b/config/src/test/java/org/hyperledger/besu/config/JsonBftConfigOptionsTest.java
index 5f530cebb5b..bbb4c4a6ceb 100644
--- a/config/src/test/java/org/hyperledger/besu/config/JsonBftConfigOptionsTest.java
+++ b/config/src/test/java/org/hyperledger/besu/config/JsonBftConfigOptionsTest.java
@@ -17,6 +17,7 @@
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import org.hyperledger.besu.datatypes.Address;
@@ -30,6 +31,7 @@ public class JsonBftConfigOptionsTest {
 
   private static final int EXPECTED_DEFAULT_EPOCH_LENGTH = 30_000;
   private static final int EXPECTED_DEFAULT_BLOCK_PERIOD = 1;
+  private static final int EXPECTED_EMPTY_DEFAULT_BLOCK_PERIOD = 0;
   private static final int EXPECTED_DEFAULT_REQUEST_TIMEOUT = 1;
   private static final int EXPECTED_DEFAULT_GOSSIPED_HISTORY_LIMIT = 1000;
   private static final int EXPECTED_DEFAULT_MESSAGE_QUEUE_LIMIT = 1000;
@@ -61,18 +63,37 @@ public void shouldGetBlockPeriodFromConfig() {
     assertThat(config.getBlockPeriodSeconds()).isEqualTo(5);
   }
 
+  @Test
+  public void shouldGetEmptyBlockPeriodFromConfig() {
+    final BftConfigOptions config = fromConfigOptions(singletonMap("xemptyblockperiodseconds", 60));
+    assertThat(config.getEmptyBlockPeriodSeconds()).isEqualTo(60);
+  }
+
   @Test
   public void shouldFallbackToDefaultBlockPeriod() {
     final BftConfigOptions config = fromConfigOptions(emptyMap());
     assertThat(config.getBlockPeriodSeconds()).isEqualTo(EXPECTED_DEFAULT_BLOCK_PERIOD);
   }
 
+  @Test
+  public void shouldFallbackToEmptyDefaultBlockPeriod() {
+    final BftConfigOptions config = fromConfigOptions(emptyMap());
+    assertThat(config.getEmptyBlockPeriodSeconds()).isEqualTo(EXPECTED_EMPTY_DEFAULT_BLOCK_PERIOD);
+  }
+
   @Test
   public void shouldGetDefaultBlockPeriodFromDefaultConfig() {
     assertThat(JsonBftConfigOptions.DEFAULT.getBlockPeriodSeconds())
         .isEqualTo(EXPECTED_DEFAULT_BLOCK_PERIOD);
   }
 
+  @Test
+  public void shouldGetDefaultEmptyBlockPeriodFromDefaultConfig() {
+
+    assertThat(JsonBftConfigOptions.DEFAULT.getEmptyBlockPeriodSeconds())
+        .isEqualTo(EXPECTED_EMPTY_DEFAULT_BLOCK_PERIOD);
+  }
+
   @Test
   public void shouldThrowOnNonPositiveBlockPeriod() {
     final BftConfigOptions config = fromConfigOptions(singletonMap("blockperiodseconds", -1));
@@ -80,6 +101,13 @@ public void shouldThrowOnNonPositiveBlockPeriod() {
         .isInstanceOf(IllegalArgumentException.class);
   }
 
+  @Test
+  public void shouldNotThrowOnNonPositiveEmptyBlockPeriod() {
+    // can be 0 to be compatible with older versions
+    final BftConfigOptions config = fromConfigOptions(singletonMap("xemptyblockperiodseconds", 0));
+    assertThatCode(() -> config.getEmptyBlockPeriodSeconds()).doesNotThrowAnyException();
+  }
+
   @Test
   public void shouldGetRequestTimeoutFromConfig() {
     final BftConfigOptions config = fromConfigOptions(singletonMap("requesttimeoutseconds", 5));
diff --git a/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BlockTimer.java b/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BlockTimer.java
index 49ef94e008c..8b98a2ddec5 100644
--- a/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BlockTimer.java
+++ b/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/BlockTimer.java
@@ -37,6 +37,8 @@ public class BlockTimer {
   private Optional<ScheduledFuture<?>> currentTimerTask;
   private final BftEventQueue queue;
   private final Clock clock;
+  private long blockPeriodSeconds;
+  private long emptyBlockPeriodSeconds;
 
   /**
    * Construct a BlockTimer with primed executor service ready to start timers
@@ -56,6 +58,8 @@ public BlockTimer(
     this.bftExecutors = bftExecutors;
     this.currentTimerTask = Optional.empty();
     this.clock = clock;
+    this.blockPeriodSeconds = 0;
+    this.emptyBlockPeriodSeconds = 0;
   }
 
   /** Cancels the current running round timer if there is one */
@@ -83,13 +87,11 @@ public synchronized void startTimer(
       final ConsensusRoundIdentifier round, final BlockHeader chainHeadHeader) {
     cancelTimer();
 
-    final long now = clock.millis();
     final long expiryTime;
 
     // Experimental option for test scenarios only. Not for production use.
     final long blockPeriodMilliseconds =
         forksSchedule.getFork(round.getSequenceNumber()).getValue().getBlockPeriodMilliseconds();
-
     if (blockPeriodMilliseconds > 0) {
       // Experimental mode for setting < 1 second block periods e.g. for CI/CD pipelines
       // running tests against Besu
@@ -99,12 +101,60 @@ public synchronized void startTimer(
           blockPeriodMilliseconds);
     } else {
       // absolute time when the timer is supposed to expire
-      final int blockPeriodSeconds =
+      final int currentBlockPeriodSeconds =
           forksSchedule.getFork(round.getSequenceNumber()).getValue().getBlockPeriodSeconds();
-      final long minimumTimeBetweenBlocksMillis = blockPeriodSeconds * 1000L;
+      final long minimumTimeBetweenBlocksMillis = currentBlockPeriodSeconds * 1000L;
       expiryTime = chainHeadHeader.getTimestamp() * 1_000 + minimumTimeBetweenBlocksMillis;
     }
 
+    setBlockTimes(round);
+
+    startTimer(round, expiryTime);
+  }
+
+  /**
+   * Checks if the empty block timer is expired
+   *
+   * @param chainHeadHeader The header of the chain head
+   * @param currentTimeInMillis The current time
+   * @return a boolean value
+   */
+  public synchronized boolean checkEmptyBlockExpired(
+      final BlockHeader chainHeadHeader, final long currentTimeInMillis) {
+    final long emptyBlockPeriodExpiryTime =
+        (chainHeadHeader.getTimestamp() + emptyBlockPeriodSeconds) * 1000;
+
+    if (currentTimeInMillis > emptyBlockPeriodExpiryTime) {
+      LOG.debug("Empty Block expired");
+      return true;
+    }
+    LOG.debug("Empty Block NOT expired");
+    return false;
+  }
+
+  /**
+   * Resets the empty block timer
+   *
+   * @param roundIdentifier The current round identifier
+   * @param chainHeadHeader The header of the chain head
+   * @param currentTimeInMillis The current time
+   */
+  public void resetTimerForEmptyBlock(
+      final ConsensusRoundIdentifier roundIdentifier,
+      final BlockHeader chainHeadHeader,
+      final long currentTimeInMillis) {
+    final long emptyBlockPeriodExpiryTime =
+        (chainHeadHeader.getTimestamp() + emptyBlockPeriodSeconds) * 1000;
+    final long nextBlockPeriodExpiryTime = currentTimeInMillis + blockPeriodSeconds * 1000;
+
+    startTimer(roundIdentifier, Math.min(emptyBlockPeriodExpiryTime, nextBlockPeriodExpiryTime));
+  }
+
+  private synchronized void startTimer(
+      final ConsensusRoundIdentifier round, final long expiryTime) {
+    cancelTimer();
+    final long now = clock.millis();
+
     if (expiryTime > now) {
       final long delay = expiryTime - now;
 
@@ -117,4 +167,29 @@ public synchronized void startTimer(
       queue.add(new BlockTimerExpiry(round));
     }
   }
+
+  private synchronized void setBlockTimes(final ConsensusRoundIdentifier round) {
+    final BftConfigOptions currentConfigOptions =
+        forksSchedule.getFork(round.getSequenceNumber()).getValue();
+    this.blockPeriodSeconds = currentConfigOptions.getBlockPeriodSeconds();
+    this.emptyBlockPeriodSeconds = currentConfigOptions.getEmptyBlockPeriodSeconds();
+  }
+
+  /**
+   * Retrieves the Block Period Seconds
+   *
+   * @return the Block Period Seconds
+   */
+  public synchronized long getBlockPeriodSeconds() {
+    return blockPeriodSeconds;
+  }
+
+  /**
+   * Retrieves the Empty Block Period Seconds
+   *
+   * @return the Empty Block Period Seconds
+   */
+  public synchronized long getEmptyBlockPeriodSeconds() {
+    return emptyBlockPeriodSeconds;
+  }
 }
diff --git a/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/MutableBftConfigOptions.java b/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/MutableBftConfigOptions.java
index 7b27c7b4c44..4cf00acd3b3 100644
--- a/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/MutableBftConfigOptions.java
+++ b/consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/MutableBftConfigOptions.java
@@ -31,6 +31,7 @@
 public class MutableBftConfigOptions implements BftConfigOptions {
   private long epochLength;
   private int blockPeriodSeconds;
+  private int emptyBlockPeriodSeconds;
   private long blockPeriodMilliseconds;
   private int requestTimeoutSeconds;
   private int gossipedHistoryLimit;
@@ -49,6 +50,7 @@ public class MutableBftConfigOptions implements BftConfigOptions {
   public MutableBftConfigOptions(final BftConfigOptions bftConfigOptions) {
     this.epochLength = bftConfigOptions.getEpochLength();
     this.blockPeriodSeconds = bftConfigOptions.getBlockPeriodSeconds();
+    this.emptyBlockPeriodSeconds = bftConfigOptions.getEmptyBlockPeriodSeconds();
     this.blockPeriodMilliseconds = bftConfigOptions.getBlockPeriodMilliseconds();
     this.requestTimeoutSeconds = bftConfigOptions.getRequestTimeoutSeconds();
     this.gossipedHistoryLimit = bftConfigOptions.getGossipedHistoryLimit();
@@ -70,6 +72,11 @@ public int getBlockPeriodSeconds() {
     return blockPeriodSeconds;
   }
 
+  @Override
+  public int getEmptyBlockPeriodSeconds() {
+    return emptyBlockPeriodSeconds;
+  }
+
   @Override
   public long getBlockPeriodMilliseconds() {
     return blockPeriodMilliseconds;
@@ -138,6 +145,15 @@ public void setBlockPeriodSeconds(final int blockPeriodSeconds) {
     this.blockPeriodSeconds = blockPeriodSeconds;
   }
 
+  /**
+   * Sets empty block period seconds.
+   *
+   * @param emptyBlockPeriodSeconds the empty block period seconds
+   */
+  public void setEmptyBlockPeriodSeconds(final int emptyBlockPeriodSeconds) {
+    this.emptyBlockPeriodSeconds = emptyBlockPeriodSeconds;
+  }
+
   /**
    * Sets block period milliseconds. Experimental for test scenarios. Not for use on production
    * systems.
diff --git a/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/ForksScheduleFactoryTest.java b/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/ForksScheduleFactoryTest.java
index 0cc9c129451..89e7330c4c4 100644
--- a/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/ForksScheduleFactoryTest.java
+++ b/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/ForksScheduleFactoryTest.java
@@ -37,7 +37,7 @@ public class ForksScheduleFactoryTest {
   @SuppressWarnings("unchecked")
   public void throwsErrorIfHasForkForGenesisBlock() {
     final BftConfigOptions genesisConfigOptions = JsonBftConfigOptions.DEFAULT;
-    final BftFork fork = createFork(0, 10);
+    final BftFork fork = createFork(0, 10, 30);
     final SpecCreator<BftConfigOptions, BftFork> specCreator = Mockito.mock(SpecCreator.class);
 
     assertThatThrownBy(
@@ -49,9 +49,9 @@ public void throwsErrorIfHasForkForGenesisBlock() {
   @SuppressWarnings("unchecked")
   public void throwsErrorIfHasForksWithDuplicateBlock() {
     final BftConfigOptions genesisConfigOptions = JsonBftConfigOptions.DEFAULT;
-    final BftFork fork1 = createFork(1, 10);
-    final BftFork fork2 = createFork(1, 20);
-    final BftFork fork3 = createFork(2, 30);
+    final BftFork fork1 = createFork(1, 10, 30);
+    final BftFork fork2 = createFork(1, 20, 60);
+    final BftFork fork3 = createFork(2, 30, 90);
     final SpecCreator<BftConfigOptions, BftFork> specCreator = Mockito.mock(SpecCreator.class);
 
     assertThatThrownBy(
@@ -66,12 +66,12 @@ public void throwsErrorIfHasForksWithDuplicateBlock() {
   public void createsScheduleUsingSpecCreator() {
     final BftConfigOptions genesisConfigOptions = JsonBftConfigOptions.DEFAULT;
     final ForkSpec<BftConfigOptions> genesisForkSpec = new ForkSpec<>(0, genesisConfigOptions);
-    final BftFork fork1 = createFork(1, 10);
-    final BftFork fork2 = createFork(2, 20);
+    final BftFork fork1 = createFork(1, 10, 20);
+    final BftFork fork2 = createFork(2, 20, 40);
     final SpecCreator<BftConfigOptions, BftFork> specCreator = Mockito.mock(SpecCreator.class);
 
-    final BftConfigOptions configOptions1 = createBftConfigOptions(10);
-    final BftConfigOptions configOptions2 = createBftConfigOptions(20);
+    final BftConfigOptions configOptions1 = createBftConfigOptions(10, 30);
+    final BftConfigOptions configOptions2 = createBftConfigOptions(20, 60);
     when(specCreator.create(genesisForkSpec, fork1)).thenReturn(configOptions1);
     when(specCreator.create(new ForkSpec<>(1, configOptions1), fork2)).thenReturn(configOptions2);
 
@@ -82,18 +82,25 @@ public void createsScheduleUsingSpecCreator() {
     assertThat(schedule.getFork(2)).isEqualTo(new ForkSpec<>(2, configOptions2));
   }
 
-  private MutableBftConfigOptions createBftConfigOptions(final int blockPeriodSeconds) {
+  private MutableBftConfigOptions createBftConfigOptions(
+      final int blockPeriodSeconds, final int emptyBlockPeriodSeconds) {
     final MutableBftConfigOptions bftConfigOptions =
         new MutableBftConfigOptions(JsonBftConfigOptions.DEFAULT);
     bftConfigOptions.setBlockPeriodSeconds(blockPeriodSeconds);
+    bftConfigOptions.setEmptyBlockPeriodSeconds(emptyBlockPeriodSeconds);
     return bftConfigOptions;
   }
 
-  private BftFork createFork(final long block, final long blockPeriodSeconds) {
+  private BftFork createFork(
+      final long block, final long blockPeriodSeconds, final long emptyBlockPeriodSeconds) {
     return new BftFork(
         JsonUtil.objectNodeFromMap(
             Map.of(
-                BftFork.FORK_BLOCK_KEY, block,
-                BftFork.BLOCK_PERIOD_SECONDS_KEY, blockPeriodSeconds)));
+                BftFork.FORK_BLOCK_KEY,
+                block,
+                BftFork.BLOCK_PERIOD_SECONDS_KEY,
+                blockPeriodSeconds,
+                BftFork.EMPTY_BLOCK_PERIOD_SECONDS_KEY,
+                emptyBlockPeriodSeconds)));
   }
 }
diff --git a/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/bft/BlockTimerTest.java b/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/bft/BlockTimerTest.java
index 70187a4d40d..7521e5eb301 100644
--- a/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/bft/BlockTimerTest.java
+++ b/consensus/common/src/test/java/org/hyperledger/besu/consensus/common/bft/BlockTimerTest.java
@@ -75,12 +75,18 @@ public void cancelTimerCancelsWhenNoTimer() {
   @Test
   public void startTimerSchedulesCorrectlyWhenExpiryIsInTheFuture() {
     final int MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS = 15;
+    final int MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS = 60;
     final long NOW_MILLIS = 505_000L;
     final long BLOCK_TIME_STAMP = 500L;
     final long EXPECTED_DELAY = 10_000L;
 
     when(mockForksSchedule.getFork(anyLong()))
-        .thenReturn(new ForkSpec<>(0, createBftFork(MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS)));
+        .thenReturn(
+            new ForkSpec<>(
+                0,
+                createBftFork(
+                    MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS,
+                    MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS)));
 
     final BlockTimer timer = new BlockTimer(mockQueue, mockForksSchedule, bftExecutors, mockClock);
 
@@ -104,12 +110,18 @@ public void startTimerSchedulesCorrectlyWhenExpiryIsInTheFuture() {
   @Test
   public void aBlockTimerExpiryEventIsAddedToTheQueueOnExpiry() throws InterruptedException {
     final int MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS = 1;
+    final int MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS = 10;
     final long NOW_MILLIS = 300_500L;
     final long BLOCK_TIME_STAMP = 300;
     final long EXPECTED_DELAY = 500;
 
     when(mockForksSchedule.getFork(anyLong()))
-        .thenReturn(new ForkSpec<>(0, createBftFork(MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS)));
+        .thenReturn(
+            new ForkSpec<>(
+                0,
+                createBftFork(
+                    MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS,
+                    MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS)));
     when(mockClock.millis()).thenReturn(NOW_MILLIS);
 
     final BlockHeader header =
@@ -149,11 +161,17 @@ public void aBlockTimerExpiryEventIsAddedToTheQueueOnExpiry() throws Interrupted
   @Test
   public void eventIsImmediatelyAddedToTheQueueIfAbsoluteExpiryIsEqualToNow() {
     final int MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS = 15;
+    final int MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS = 60;
     final long NOW_MILLIS = 515_000L;
     final long BLOCK_TIME_STAMP = 500;
 
     when(mockForksSchedule.getFork(anyLong()))
-        .thenReturn(new ForkSpec<>(0, createBftFork(MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS)));
+        .thenReturn(
+            new ForkSpec<>(
+                0,
+                createBftFork(
+                    MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS,
+                    MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS)));
 
     final BlockTimer timer = new BlockTimer(mockQueue, mockForksSchedule, bftExecutors, mockClock);
 
@@ -179,11 +197,17 @@ public void eventIsImmediatelyAddedToTheQueueIfAbsoluteExpiryIsEqualToNow() {
   @Test
   public void eventIsImmediatelyAddedToTheQueueIfAbsoluteExpiryIsInThePast() {
     final int MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS = 15;
+    final int MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS = 60;
     final long NOW_MILLIS = 520_000L;
     final long BLOCK_TIME_STAMP = 500L;
 
     when(mockForksSchedule.getFork(anyLong()))
-        .thenReturn(new ForkSpec<>(0, createBftFork(MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS)));
+        .thenReturn(
+            new ForkSpec<>(
+                0,
+                createBftFork(
+                    MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS,
+                    MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS)));
 
     final BlockTimer timer = new BlockTimer(mockQueue, mockForksSchedule, bftExecutors, mockClock);
 
@@ -209,11 +233,17 @@ public void eventIsImmediatelyAddedToTheQueueIfAbsoluteExpiryIsInThePast() {
   @Test
   public void startTimerCancelsExistingTimer() {
     final int MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS = 15;
+    final int MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS = 60;
     final long NOW_MILLIS = 500_000L;
     final long BLOCK_TIME_STAMP = 500L;
 
     when(mockForksSchedule.getFork(anyLong()))
-        .thenReturn(new ForkSpec<>(0, createBftFork(MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS)));
+        .thenReturn(
+            new ForkSpec<>(
+                0,
+                createBftFork(
+                    MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS,
+                    MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS)));
 
     final BlockTimer timer = new BlockTimer(mockQueue, mockForksSchedule, bftExecutors, mockClock);
 
@@ -237,11 +267,17 @@ public void startTimerCancelsExistingTimer() {
   @Test
   public void runningFollowsTheStateOfTheTimer() {
     final int MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS = 15;
+    final int MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS = 60;
     final long NOW_MILLIS = 500_000L;
     final long BLOCK_TIME_STAMP = 500L;
 
     when(mockForksSchedule.getFork(anyLong()))
-        .thenReturn(new ForkSpec<>(0, createBftFork(MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS)));
+        .thenReturn(
+            new ForkSpec<>(
+                0,
+                createBftFork(
+                    MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS,
+                    MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS)));
 
     final BlockTimer timer = new BlockTimer(mockQueue, mockForksSchedule, bftExecutors, mockClock);
 
@@ -263,10 +299,42 @@ public void runningFollowsTheStateOfTheTimer() {
     assertThat(timer.isRunning()).isFalse();
   }
 
-  private BftConfigOptions createBftFork(final int blockPeriodSeconds) {
+  @Test
+  public void checkBlockTimerEmptyAndNonEmptyPeriodSecods() {
+    final int MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS = 15;
+    final int MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS = 60;
+    final long BLOCK_TIME_STAMP = 500L;
+    final ConsensusRoundIdentifier round =
+        new ConsensusRoundIdentifier(0xFEDBCA9876543210L, 0x12345678);
+    final BlockHeader header =
+        new BlockHeaderTestFixture().timestamp(BLOCK_TIME_STAMP).buildHeader();
+    final ScheduledFuture<?> mockedFuture = mock(ScheduledFuture.class);
+    Mockito.<ScheduledFuture<?>>when(
+            bftExecutors.scheduleTask(any(Runnable.class), anyLong(), any()))
+        .thenReturn(mockedFuture);
+
+    when(mockForksSchedule.getFork(anyLong()))
+        .thenReturn(
+            new ForkSpec<>(
+                0,
+                createBftFork(
+                    MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS,
+                    MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS)));
+
+    final BlockTimer timer = new BlockTimer(mockQueue, mockForksSchedule, bftExecutors, mockClock);
+    timer.startTimer(round, header);
+
+    assertThat(timer.getBlockPeriodSeconds()).isEqualTo(MINIMAL_TIME_BETWEEN_BLOCKS_SECONDS);
+    assertThat(timer.getEmptyBlockPeriodSeconds())
+        .isEqualTo(MINIMAL_TIME_BETWEEN_EMPTY_BLOCKS_SECONDS);
+  }
+
+  private BftConfigOptions createBftFork(
+      final int blockPeriodSeconds, final int emptyBlockPeriodSeconds) {
     final MutableBftConfigOptions bftConfigOptions =
         new MutableBftConfigOptions(JsonBftConfigOptions.DEFAULT);
     bftConfigOptions.setBlockPeriodSeconds(blockPeriodSeconds);
+    bftConfigOptions.setEmptyBlockPeriodSeconds(emptyBlockPeriodSeconds);
     return bftConfigOptions;
   }
 }
diff --git a/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/QbftForksSchedulesFactory.java b/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/QbftForksSchedulesFactory.java
index 0f0d467de37..be8e2ed5c46 100644
--- a/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/QbftForksSchedulesFactory.java
+++ b/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/QbftForksSchedulesFactory.java
@@ -49,6 +49,7 @@ private static QbftConfigOptions createQbftConfigOptions(
         new MutableQbftConfigOptions(lastSpec.getValue());
 
     fork.getBlockPeriodSeconds().ifPresent(bftConfigOptions::setBlockPeriodSeconds);
+    fork.getEmptyBlockPeriodSeconds().ifPresent(bftConfigOptions::setEmptyBlockPeriodSeconds);
     fork.getBlockPeriodMilliseconds().ifPresent(bftConfigOptions::setBlockPeriodMilliseconds);
     fork.getBlockRewardWei().ifPresent(bftConfigOptions::setBlockRewardWei);
 
diff --git a/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManager.java b/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManager.java
index f79537cc0f9..757998b5ca6 100644
--- a/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManager.java
+++ b/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManager.java
@@ -28,6 +28,7 @@
 import org.hyperledger.besu.consensus.qbft.validation.FutureRoundProposalMessageValidator;
 import org.hyperledger.besu.consensus.qbft.validation.MessageValidatorFactory;
 import org.hyperledger.besu.datatypes.Address;
+import org.hyperledger.besu.ethereum.core.Block;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.plugin.services.securitymodule.SecurityModuleException;
 
@@ -130,19 +131,57 @@ public void handleBlockTimerExpiry(final ConsensusRoundIdentifier roundIdentifie
 
     logValidatorChanges(qbftRound);
 
+    if (roundIdentifier.equals(qbftRound.getRoundIdentifier())) {
+      buildBlockAndMaybePropose(roundIdentifier, qbftRound);
+    } else {
+      LOG.trace(
+          "Block timer expired for a round ({}) other than current ({})",
+          roundIdentifier,
+          qbftRound.getRoundIdentifier());
+    }
+  }
+
+  private void buildBlockAndMaybePropose(
+      final ConsensusRoundIdentifier roundIdentifier, final QbftRound qbftRound) {
+
     // mining will be checked against round 0 as the current round is initialised to 0 above
     final boolean isProposer =
         finalState.isLocalNodeProposerForRound(qbftRound.getRoundIdentifier());
 
-    if (isProposer) {
-      if (roundIdentifier.equals(qbftRound.getRoundIdentifier())) {
-        final long headerTimeStampSeconds = Math.round(clock.millis() / 1000D);
-        qbftRound.createAndSendProposalMessage(headerTimeStampSeconds);
+    if (!isProposer) {
+      // nothing to do here...
+      LOG.trace("This node is not a proposer so it will not send a proposal: " + roundIdentifier);
+      return;
+    }
+
+    final long headerTimeStampSeconds = Math.round(clock.millis() / 1000D);
+    final Block block = qbftRound.createBlock(headerTimeStampSeconds);
+    final boolean blockHasTransactions = !block.getBody().getTransactions().isEmpty();
+    if (blockHasTransactions) {
+      LOG.trace(
+          "Block has transactions and this node is a proposer so it will send a proposal: "
+              + roundIdentifier);
+      qbftRound.updateStateWithProposalAndTransmit(block);
+
+    } else {
+      // handle the block times period
+      final long currentTimeInMillis = finalState.getClock().millis();
+      boolean emptyBlockExpired =
+          finalState.getBlockTimer().checkEmptyBlockExpired(parentHeader, currentTimeInMillis);
+      if (emptyBlockExpired) {
+        LOG.trace(
+            "Block has no transactions and this node is a proposer so it will send a proposal: "
+                + roundIdentifier);
+        qbftRound.updateStateWithProposalAndTransmit(block);
       } else {
         LOG.trace(
-            "Block timer expired for a round ({}) other than current ({})",
-            roundIdentifier,
-            qbftRound.getRoundIdentifier());
+            "Block has no transactions but emptyBlockPeriodSeconds did not expired yet: "
+                + roundIdentifier);
+        finalState
+            .getBlockTimer()
+            .resetTimerForEmptyBlock(roundIdentifier, parentHeader, currentTimeInMillis);
+        finalState.getRoundTimer().cancelTimer();
+        currentRound = Optional.empty();
       }
     }
   }
diff --git a/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRound.java b/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRound.java
index f7c43ef43dd..e0a8cb1c726 100644
--- a/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRound.java
+++ b/consensus/qbft/src/main/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRound.java
@@ -132,17 +132,14 @@ public ConsensusRoundIdentifier getRoundIdentifier() {
   }
 
   /**
-   * Create and send proposal message.
+   * Create a block
    *
-   * @param headerTimeStampSeconds the header time stamp seconds
+   * @param headerTimeStampSeconds of the block
+   * @return a Block
    */
-  public void createAndSendProposalMessage(final long headerTimeStampSeconds) {
+  public Block createBlock(final long headerTimeStampSeconds) {
     LOG.debug("Creating proposed block. round={}", roundState.getRoundIdentifier());
-    final Block block =
-        blockCreator.createBlock(headerTimeStampSeconds, this.parentHeader).getBlock();
-
-    LOG.trace("Creating proposed block blockHeader={}", block.getHeader());
-    updateStateWithProposalAndTransmit(block, emptyList(), emptyList());
+    return blockCreator.createBlock(headerTimeStampSeconds, this.parentHeader).getBlock();
   }
 
   /**
@@ -172,6 +169,15 @@ public void startRoundWith(
         bestPreparedCertificate.map(PreparedCertificate::getPrepares).orElse(emptyList()));
   }
 
+  /**
+   * Update state with proposal and transmit.
+   *
+   * @param block the block
+   */
+  protected void updateStateWithProposalAndTransmit(final Block block) {
+    updateStateWithProposalAndTransmit(block, emptyList(), emptyList());
+  }
+
   /**
    * Update state with proposal and transmit.
    *
diff --git a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/MutableQbftConfigOptionsTest.java b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/MutableQbftConfigOptionsTest.java
index 8ad39adc5b5..2b04796b1a2 100644
--- a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/MutableQbftConfigOptionsTest.java
+++ b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/MutableQbftConfigOptionsTest.java
@@ -37,4 +37,24 @@ public void getValidatorContractAddressNormalization() {
 
     assertThat(mutableQbftConfigOptions.getValidatorContractAddress()).hasValue("0xabc");
   }
+
+  @Test
+  public void checkBlockPeriodSeconds() {
+    when(qbftConfigOptions.getBlockPeriodSeconds()).thenReturn(2);
+
+    final MutableQbftConfigOptions mutableQbftConfigOptions =
+        new MutableQbftConfigOptions(qbftConfigOptions);
+
+    assertThat(mutableQbftConfigOptions.getBlockPeriodSeconds()).isEqualTo(2);
+  }
+
+  @Test
+  public void checkEmptyBlockPeriodSeconds() {
+    when(qbftConfigOptions.getEmptyBlockPeriodSeconds()).thenReturn(60);
+
+    final MutableQbftConfigOptions mutableQbftConfigOptions =
+        new MutableQbftConfigOptions(qbftConfigOptions);
+
+    assertThat(mutableQbftConfigOptions.getEmptyBlockPeriodSeconds()).isEqualTo(60);
+  }
 }
diff --git a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManagerTest.java b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManagerTest.java
index 171b1607f39..190c87e020f 100644
--- a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManagerTest.java
+++ b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftBlockHeightManagerTest.java
@@ -157,8 +157,10 @@ public void setup() {
     when(messageValidator.validateCommit(any())).thenReturn(true);
     when(messageValidator.validatePrepare(any())).thenReturn(true);
     when(finalState.getBlockTimer()).thenReturn(blockTimer);
+    when(finalState.getRoundTimer()).thenReturn(roundTimer);
     when(finalState.getQuorum()).thenReturn(3);
     when(finalState.getValidatorMulticaster()).thenReturn(validatorMulticaster);
+    when(finalState.getClock()).thenReturn(clock);
     when(blockCreator.createBlock(anyLong(), any()))
         .thenReturn(
             new BlockCreationResult(
@@ -267,6 +269,7 @@ public void doesNotStartRoundTimerOnStart() {
   @Test
   public void onBlockTimerExpiryRoundTimerIsStartedAndProposalMessageIsTransmitted() {
     when(finalState.isLocalNodeProposerForRound(roundIdentifier)).thenReturn(true);
+    when(blockTimer.checkEmptyBlockExpired(any(), eq(0l))).thenReturn(true);
 
     final QbftBlockHeightManager manager =
         new QbftBlockHeightManager(
@@ -290,6 +293,7 @@ public void onBlockTimerExpiryRoundTimerIsStartedAndProposalMessageIsTransmitted
   public void
       onBlockTimerExpiryForNonProposerRoundTimerIsStartedAndNoProposalMessageIsTransmitted() {
     when(finalState.isLocalNodeProposerForRound(roundIdentifier)).thenReturn(false);
+    when(blockTimer.checkEmptyBlockExpired(any(), eq(0l))).thenReturn(true);
 
     final QbftBlockHeightManager manager =
         new QbftBlockHeightManager(
@@ -463,6 +467,7 @@ public void messagesForFutureRoundsAreBufferedAndUsedToPreloadNewRoundWhenItIsSt
   public void messagesForCurrentRoundAreBufferedAndUsedToPreloadRoundWhenItIsStarted() {
     when(finalState.getQuorum()).thenReturn(1);
     when(finalState.isLocalNodeProposerForRound(roundIdentifier)).thenReturn(true);
+    when(blockTimer.checkEmptyBlockExpired(any(), eq(0l))).thenReturn(true);
 
     final QbftBlockHeightManager manager =
         new QbftBlockHeightManager(
@@ -500,6 +505,7 @@ public void messagesForCurrentRoundAreBufferedAndUsedToPreloadRoundWhenItIsStart
   @Test
   public void preparedCertificateIncludedInRoundChangeMessageOnRoundTimeoutExpired() {
     when(finalState.isLocalNodeProposerForRound(any())).thenReturn(true);
+    when(blockTimer.checkEmptyBlockExpired(any(), eq(0l))).thenReturn(true);
 
     final QbftBlockHeightManager manager =
         new QbftBlockHeightManager(
@@ -577,4 +583,24 @@ public void illegalFutureRoundProposalDoesNotTriggerNewRound() {
     manager.handleProposalPayload(futureRoundProposal);
     verify(roundFactory, never()).createNewRound(any(), anyInt());
   }
+
+  @Test
+  public void checkOnlyEmptyBlockPeriodSecondsIsInvokedForBlocksWithNoTransactions() {
+    when(finalState.isLocalNodeProposerForRound(roundIdentifier)).thenReturn(true);
+
+    final QbftBlockHeightManager manager =
+        new QbftBlockHeightManager(
+            headerTestFixture.buildHeader(),
+            finalState,
+            roundChangeManager,
+            roundFactory,
+            clock,
+            messageValidatorFactory,
+            messageFactory);
+
+    manager.handleBlockTimerExpiry(roundIdentifier);
+
+    verify(blockTimer, times(0)).getEmptyBlockPeriodSeconds();
+    verify(blockTimer, times(0)).getBlockPeriodSeconds();
+  }
 }
diff --git a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRoundTest.java b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRoundTest.java
index 514fd653a1d..fb84c1c38fa 100644
--- a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRoundTest.java
+++ b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/statemachine/QbftRoundTest.java
@@ -29,7 +29,6 @@
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
-import org.hyperledger.besu.consensus.common.bft.BftBlockHashing;
 import org.hyperledger.besu.consensus.common.bft.BftContext;
 import org.hyperledger.besu.consensus.common.bft.BftExtraData;
 import org.hyperledger.besu.consensus.common.bft.BftExtraDataCodec;
@@ -48,7 +47,6 @@
 import org.hyperledger.besu.crypto.SignatureAlgorithmFactory;
 import org.hyperledger.besu.cryptoservices.NodeKey;
 import org.hyperledger.besu.cryptoservices.NodeKeyUtils;
-import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.ProtocolContext;
 import org.hyperledger.besu.ethereum.blockcreation.BlockCreationTiming;
 import org.hyperledger.besu.ethereum.blockcreation.BlockCreator.BlockCreationResult;
@@ -196,90 +194,6 @@ public void onReceptionOfValidProposalSendsAPrepareToNetworkPeers() {
     verify(transmitter, never()).multicastCommit(any(), any(), any());
   }
 
-  @Test
-  public void sendsAProposalAndPrepareWhenSendProposalRequested() {
-    final RoundState roundState = new RoundState(roundIdentifier, 3, messageValidator);
-    final QbftRound round =
-        new QbftRound(
-            roundState,
-            blockCreator,
-            protocolContext,
-            protocolSchedule,
-            subscribers,
-            nodeKey,
-            messageFactory,
-            transmitter,
-            roundTimer,
-            bftExtraDataCodec,
-            parentHeader);
-
-    round.createAndSendProposalMessage(15);
-    verify(transmitter, times(1))
-        .multicastProposal(
-            roundIdentifier, proposedBlock, Collections.emptyList(), Collections.emptyList());
-    verify(transmitter, times(1)).multicastPrepare(roundIdentifier, proposedBlock.getHash());
-    verify(transmitter, never()).multicastCommit(any(), any(), any());
-  }
-
-  @Test
-  public void singleValidatorImportBlocksImmediatelyOnProposalCreation() {
-    final RoundState roundState = new RoundState(roundIdentifier, 1, messageValidator);
-    final QbftRound round =
-        new QbftRound(
-            roundState,
-            blockCreator,
-            protocolContext,
-            protocolSchedule,
-            subscribers,
-            nodeKey,
-            messageFactory,
-            transmitter,
-            roundTimer,
-            bftExtraDataCodec,
-            parentHeader);
-    round.createAndSendProposalMessage(15);
-    verify(transmitter, times(1))
-        .multicastProposal(
-            roundIdentifier, proposedBlock, Collections.emptyList(), Collections.emptyList());
-    verify(transmitter, times(1)).multicastPrepare(roundIdentifier, proposedBlock.getHash());
-    verify(transmitter, times(1)).multicastCommit(any(), any(), any());
-  }
-
-  @Test
-  public void localNodeProposesToNetworkOfTwoValidatorsImportsOnReceptionOfCommitFromPeer() {
-    final RoundState roundState = new RoundState(roundIdentifier, 2, messageValidator);
-    final QbftRound round =
-        new QbftRound(
-            roundState,
-            blockCreator,
-            protocolContext,
-            protocolSchedule,
-            subscribers,
-            nodeKey,
-            messageFactory,
-            transmitter,
-            roundTimer,
-            bftExtraDataCodec,
-            parentHeader);
-
-    final Hash commitSealHash =
-        new BftBlockHashing(new QbftExtraDataCodec())
-            .calculateDataHashForCommittedSeal(proposedBlock.getHeader(), proposedExtraData);
-    final SECPSignature localCommitSeal = nodeKey.sign(commitSealHash);
-
-    round.createAndSendProposalMessage(15);
-    verify(transmitter, never()).multicastCommit(any(), any(), any());
-
-    round.handlePrepareMessage(
-        messageFactory2.createPrepare(roundIdentifier, proposedBlock.getHash()));
-
-    verify(transmitter, times(1))
-        .multicastCommit(roundIdentifier, proposedBlock.getHash(), localCommitSeal);
-
-    round.handleCommitMessage(
-        messageFactory.createCommit(roundIdentifier, proposedBlock.getHash(), remoteCommitSeal));
-  }
-
   @Test
   public void aProposalWithAnewBlockIsSentUponReceptionOfARoundChangeWithNoCertificate() {
     final RoundState roundState = new RoundState(roundIdentifier, 2, messageValidator);
@@ -393,26 +307,6 @@ public void creatingNewBlockFromEmptyPreparedCertificateUpdatesInternalState() {
     assertThat(roundState.isPrepared()).isTrue();
   }
 
-  @Test
-  public void creatingNewBlockNotifiesBlockMiningObservers() {
-    final RoundState roundState = new RoundState(roundIdentifier, 1, messageValidator);
-    final QbftRound round =
-        new QbftRound(
-            roundState,
-            blockCreator,
-            protocolContext,
-            protocolSchedule,
-            subscribers,
-            nodeKey,
-            messageFactory,
-            transmitter,
-            roundTimer,
-            bftExtraDataCodec,
-            parentHeader);
-    round.createAndSendProposalMessage(15);
-    verify(minedBlockObserver).blockMined(any());
-  }
-
   @Test
   public void blockIsOnlyImportedOnceWhenCommitsAreReceivedBeforeProposal() {
     final ConsensusRoundIdentifier roundIdentifier = new ConsensusRoundIdentifier(1, 0);
diff --git a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/validator/QbftForksSchedulesFactoryTest.java b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/validator/QbftForksSchedulesFactoryTest.java
index d57ad8dd9fa..9622f98074a 100644
--- a/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/validator/QbftForksSchedulesFactoryTest.java
+++ b/consensus/qbft/src/test/java/org/hyperledger/besu/consensus/qbft/validator/QbftForksSchedulesFactoryTest.java
@@ -63,6 +63,8 @@ public void createsScheduleWithForkThatOverridesGenesisValues() {
                 List.of("1", "2", "3"),
                 BftFork.BLOCK_PERIOD_SECONDS_KEY,
                 10,
+                BftFork.EMPTY_BLOCK_PERIOD_SECONDS_KEY,
+                60,
                 BftFork.BLOCK_REWARD_KEY,
                 "5",
                 QbftFork.VALIDATOR_SELECTION_MODE_KEY,
@@ -78,6 +80,7 @@ public void createsScheduleWithForkThatOverridesGenesisValues() {
 
     final Map<String, Object> forkOptions = new HashMap<>(configOptions.asMap());
     forkOptions.put(BftFork.BLOCK_PERIOD_SECONDS_KEY, 10);
+    forkOptions.put(BftFork.EMPTY_BLOCK_PERIOD_SECONDS_KEY, 60);
     forkOptions.put(BftFork.BLOCK_REWARD_KEY, "5");
     forkOptions.put(QbftFork.VALIDATOR_SELECTION_MODE_KEY, "5");
     forkOptions.put(QbftFork.VALIDATOR_CONTRACT_ADDRESS_KEY, "10");
diff --git a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/MiningParameters.java b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/MiningParameters.java
index 20a8a31084d..1921b3568b1 100644
--- a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/MiningParameters.java
+++ b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/MiningParameters.java
@@ -131,6 +131,11 @@ public MiningParameters setBlockPeriodSeconds(final int blockPeriodSeconds) {
     return this;
   }
 
+  public MiningParameters setEmptyBlockPeriodSeconds(final int emptyBlockPeriodSeconds) {
+    getMutableRuntimeValues().emptyBlockPeriodSeconds = OptionalInt.of(emptyBlockPeriodSeconds);
+    return this;
+  }
+
   @Value.Default
   public boolean isStratumMiningEnabled() {
     return false;
@@ -231,6 +236,8 @@ default double getMinBlockOccupancyRatio() {
 
     OptionalInt getBlockPeriodSeconds();
 
+    OptionalInt getEmptyBlockPeriodSeconds();
+
     Optional<Address> getCoinbase();
 
     OptionalLong getTargetGasLimit();
@@ -248,6 +255,7 @@ static class MutableRuntimeValues {
     private volatile OptionalLong targetGasLimit;
     private volatile Optional<Iterable<Long>> nonceGenerator;
     private volatile OptionalInt blockPeriodSeconds;
+    private volatile OptionalInt emptyBlockPeriodSeconds;
 
     private MutableRuntimeValues(final MutableInitValues initValues) {
       miningEnabled = initValues.isMiningEnabled();
@@ -259,6 +267,7 @@ private MutableRuntimeValues(final MutableInitValues initValues) {
       targetGasLimit = initValues.getTargetGasLimit();
       nonceGenerator = initValues.nonceGenerator();
       blockPeriodSeconds = initValues.getBlockPeriodSeconds();
+      emptyBlockPeriodSeconds = initValues.getEmptyBlockPeriodSeconds();
     }
 
     @Override
@@ -274,7 +283,8 @@ public boolean equals(final Object o) {
           && Objects.equals(minPriorityFeePerGas, that.minPriorityFeePerGas)
           && Objects.equals(targetGasLimit, that.targetGasLimit)
           && Objects.equals(nonceGenerator, that.nonceGenerator)
-          && Objects.equals(blockPeriodSeconds, that.blockPeriodSeconds);
+          && Objects.equals(blockPeriodSeconds, that.blockPeriodSeconds)
+          && Objects.equals(emptyBlockPeriodSeconds, that.emptyBlockPeriodSeconds);
     }
 
     @Override
diff --git a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/plugins/PluginConfiguration.java b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/plugins/PluginConfiguration.java
index ebc99f647af..b9122888725 100644
--- a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/plugins/PluginConfiguration.java
+++ b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/core/plugins/PluginConfiguration.java
@@ -23,14 +23,17 @@ public class PluginConfiguration {
   private final List<PluginInfo> requestedPlugins;
   private final Path pluginsDir;
   private final boolean externalPluginsEnabled;
+  private final boolean continueOnPluginError;
 
   public PluginConfiguration(
       final List<PluginInfo> requestedPlugins,
       final Path pluginsDir,
-      final boolean externalPluginsEnabled) {
+      final boolean externalPluginsEnabled,
+      final boolean continueOnPluginError) {
     this.requestedPlugins = requestedPlugins;
     this.pluginsDir = pluginsDir;
     this.externalPluginsEnabled = externalPluginsEnabled;
+    this.continueOnPluginError = continueOnPluginError;
   }
 
   public List<String> getRequestedPlugins() {
@@ -47,6 +50,10 @@ public boolean isExternalPluginsEnabled() {
     return externalPluginsEnabled;
   }
 
+  public boolean isContinueOnPluginError() {
+    return continueOnPluginError;
+  }
+
   public static Path defaultPluginsDir() {
     String pluginsDirProperty = System.getProperty("besu.plugins.dir");
     return pluginsDirProperty == null
@@ -62,6 +69,7 @@ public static class Builder {
     private List<PluginInfo> requestedPlugins;
     private Path pluginsDir;
     private boolean externalPluginsEnabled = true;
+    private boolean continueOnPluginError = false;
 
     public Builder requestedPlugins(final List<PluginInfo> requestedPlugins) {
       this.requestedPlugins = requestedPlugins;
@@ -78,11 +86,17 @@ public Builder externalPluginsEnabled(final boolean externalPluginsEnabled) {
       return this;
     }
 
+    public Builder continueOnPluginError(final boolean continueOnPluginError) {
+      this.continueOnPluginError = continueOnPluginError;
+      return this;
+    }
+
     public PluginConfiguration build() {
       if (pluginsDir == null) {
         pluginsDir = PluginConfiguration.defaultPluginsDir();
       }
-      return new PluginConfiguration(requestedPlugins, pluginsDir, externalPluginsEnabled);
+      return new PluginConfiguration(
+          requestedPlugins, pluginsDir, externalPluginsEnabled, continueOnPluginError);
     }
   }
 }
diff --git a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/MainnetProtocolSpecs.java b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/MainnetProtocolSpecs.java
index e6d7a88cbdf..b5b2f678d67 100644
--- a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/MainnetProtocolSpecs.java
+++ b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/MainnetProtocolSpecs.java
@@ -14,10 +14,8 @@
  */
 package org.hyperledger.besu.ethereum.mainnet;
 
-import static org.hyperledger.besu.ethereum.mainnet.requests.DepositRequestProcessor.DEFAULT_DEPOSIT_CONTRACT_ADDRESS;
 import static org.hyperledger.besu.ethereum.mainnet.requests.MainnetRequestsValidator.pragueRequestsProcessors;
 import static org.hyperledger.besu.ethereum.mainnet.requests.MainnetRequestsValidator.pragueRequestsValidator;
-import static org.hyperledger.besu.ethereum.mainnet.requests.WithdrawalRequestProcessor.DEFAULT_WITHDRAWAL_REQUEST_CONTRACT_ADDRESS;
 
 import org.hyperledger.besu.config.GenesisConfigOptions;
 import org.hyperledger.besu.config.PowAlgorithm;
@@ -41,6 +39,7 @@
 import org.hyperledger.besu.ethereum.mainnet.feemarket.BaseFeeMarket;
 import org.hyperledger.besu.ethereum.mainnet.feemarket.FeeMarket;
 import org.hyperledger.besu.ethereum.mainnet.parallelization.MainnetParallelBlockProcessor;
+import org.hyperledger.besu.ethereum.mainnet.requests.RequestContractAddresses;
 import org.hyperledger.besu.ethereum.privacy.PrivateTransactionProcessor;
 import org.hyperledger.besu.ethereum.privacy.PrivateTransactionValidator;
 import org.hyperledger.besu.ethereum.privacy.storage.PrivateMetadataUpdater;
@@ -767,12 +766,8 @@ static ProtocolSpecBuilder pragueDefinition(
       final boolean isParallelTxProcessingEnabled,
       final MetricsSystem metricsSystem) {
 
-    final Address withdrawalRequestContractAddress =
-        genesisConfigOptions
-            .getWithdrawalRequestContractAddress()
-            .orElse(DEFAULT_WITHDRAWAL_REQUEST_CONTRACT_ADDRESS);
-    final Address depositContractAddress =
-        genesisConfigOptions.getDepositContractAddress().orElse(DEFAULT_DEPOSIT_CONTRACT_ADDRESS);
+    RequestContractAddresses requestContractAddresses =
+        RequestContractAddresses.fromGenesis(genesisConfigOptions);
 
     return cancunDefinition(
             chainId,
@@ -794,10 +789,9 @@ static ProtocolSpecBuilder pragueDefinition(
         .precompileContractRegistryBuilder(MainnetPrecompiledContractRegistries::prague)
 
         // EIP-7002 Withdrawals / EIP-6610 Deposits / EIP-7685 Requests
-        .requestsValidator(pragueRequestsValidator(depositContractAddress))
+        .requestsValidator(pragueRequestsValidator(requestContractAddresses))
         // EIP-7002 Withdrawals / EIP-6610 Deposits / EIP-7685 Requests
-        .requestProcessorCoordinator(
-            pragueRequestsProcessors(withdrawalRequestContractAddress, depositContractAddress))
+        .requestProcessorCoordinator(pragueRequestsProcessors(requestContractAddresses))
 
         // change to accept EIP-7702 transactions
         .transactionValidatorFactoryBuilder(
diff --git a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/ConsolidationRequestProcessor.java b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/ConsolidationRequestProcessor.java
index 0a48d8278c9..641720670fb 100644
--- a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/ConsolidationRequestProcessor.java
+++ b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/ConsolidationRequestProcessor.java
@@ -22,13 +22,18 @@
 
 public class ConsolidationRequestProcessor
     extends AbstractSystemCallRequestProcessor<ConsolidationRequest> {
-  public static final Address CONSOLIDATION_REQUEST_PREDEPLOY_ADDRESS =
+  public static final Address CONSOLIDATION_REQUEST_CONTRACT_ADDRESS =
       Address.fromHexString("0x00b42dbF2194e931E80326D950320f7d9Dbeac02");
 
   private static final int ADDRESS_BYTES = 20;
   private static final int PUBLIC_KEY_BYTES = 48;
   private static final int CONSOLIDATION_REQUEST_BYTES_SIZE =
       ADDRESS_BYTES + PUBLIC_KEY_BYTES + PUBLIC_KEY_BYTES;
+  private final Address consolidationRequestContractAddress;
+
+  public ConsolidationRequestProcessor(final Address consolidationRequestContractAddress) {
+    this.consolidationRequestContractAddress = consolidationRequestContractAddress;
+  }
 
   /**
    * Gets the call address for consolidation requests.
@@ -37,7 +42,7 @@ public class ConsolidationRequestProcessor
    */
   @Override
   protected Address getCallAddress() {
-    return CONSOLIDATION_REQUEST_PREDEPLOY_ADDRESS;
+    return consolidationRequestContractAddress;
   }
 
   /**
diff --git a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/MainnetRequestsValidator.java b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/MainnetRequestsValidator.java
index 6e61a0343c3..9c86d18f7ad 100644
--- a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/MainnetRequestsValidator.java
+++ b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/MainnetRequestsValidator.java
@@ -14,27 +14,34 @@
  */
 package org.hyperledger.besu.ethereum.mainnet.requests;
 
-import org.hyperledger.besu.datatypes.Address;
 import org.hyperledger.besu.datatypes.RequestType;
 
 public class MainnetRequestsValidator {
   public static RequestsValidatorCoordinator pragueRequestsValidator(
-      final Address depositContractAddress) {
+      final RequestContractAddresses requestContractAddresses) {
     return new RequestsValidatorCoordinator.Builder()
         .addValidator(RequestType.WITHDRAWAL, new WithdrawalRequestValidator())
         .addValidator(RequestType.CONSOLIDATION, new ConsolidationRequestValidator())
-        .addValidator(RequestType.DEPOSIT, new DepositRequestValidator(depositContractAddress))
+        .addValidator(
+            RequestType.DEPOSIT,
+            new DepositRequestValidator(requestContractAddresses.getDepositContractAddress()))
         .build();
   }
 
   public static RequestProcessorCoordinator pragueRequestsProcessors(
-      final Address withdrawalRequestContractAddress, final Address depositContractAddress) {
+      final RequestContractAddresses requestContractAddresses) {
     return new RequestProcessorCoordinator.Builder()
         .addProcessor(
             RequestType.WITHDRAWAL,
-            new WithdrawalRequestProcessor(withdrawalRequestContractAddress))
-        .addProcessor(RequestType.CONSOLIDATION, new ConsolidationRequestProcessor())
-        .addProcessor(RequestType.DEPOSIT, new DepositRequestProcessor(depositContractAddress))
+            new WithdrawalRequestProcessor(
+                requestContractAddresses.getWithdrawalRequestContractAddress()))
+        .addProcessor(
+            RequestType.CONSOLIDATION,
+            new ConsolidationRequestProcessor(
+                requestContractAddresses.getConsolidationRequestContractAddress()))
+        .addProcessor(
+            RequestType.DEPOSIT,
+            new DepositRequestProcessor(requestContractAddresses.getDepositContractAddress()))
         .build();
   }
 }
diff --git a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/RequestContractAddresses.java b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/RequestContractAddresses.java
new file mode 100644
index 00000000000..b75677dda79
--- /dev/null
+++ b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/mainnet/requests/RequestContractAddresses.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.ethereum.mainnet.requests;
+
+import static org.hyperledger.besu.ethereum.mainnet.requests.ConsolidationRequestProcessor.CONSOLIDATION_REQUEST_CONTRACT_ADDRESS;
+import static org.hyperledger.besu.ethereum.mainnet.requests.DepositRequestProcessor.DEFAULT_DEPOSIT_CONTRACT_ADDRESS;
+import static org.hyperledger.besu.ethereum.mainnet.requests.WithdrawalRequestProcessor.DEFAULT_WITHDRAWAL_REQUEST_CONTRACT_ADDRESS;
+
+import org.hyperledger.besu.config.GenesisConfigOptions;
+import org.hyperledger.besu.datatypes.Address;
+
+public class RequestContractAddresses {
+  private final Address withdrawalRequestContractAddress;
+  private final Address depositContractAddress;
+  private final Address consolidationRequestContractAddress;
+
+  public RequestContractAddresses(
+      final Address withdrawalRequestContractAddress,
+      final Address depositContractAddress,
+      final Address consolidationRequestContractAddress) {
+    this.withdrawalRequestContractAddress = withdrawalRequestContractAddress;
+    this.depositContractAddress = depositContractAddress;
+    this.consolidationRequestContractAddress = consolidationRequestContractAddress;
+  }
+
+  public static RequestContractAddresses fromGenesis(
+      final GenesisConfigOptions genesisConfigOptions) {
+    return new RequestContractAddresses(
+        genesisConfigOptions
+            .getWithdrawalRequestContractAddress()
+            .orElse(DEFAULT_WITHDRAWAL_REQUEST_CONTRACT_ADDRESS),
+        genesisConfigOptions.getDepositContractAddress().orElse(DEFAULT_DEPOSIT_CONTRACT_ADDRESS),
+        genesisConfigOptions
+            .getConsolidationRequestContractAddress()
+            .orElse(CONSOLIDATION_REQUEST_CONTRACT_ADDRESS));
+  }
+
+  public Address getWithdrawalRequestContractAddress() {
+    return withdrawalRequestContractAddress;
+  }
+
+  public Address getDepositContractAddress() {
+    return depositContractAddress;
+  }
+
+  public Address getConsolidationRequestContractAddress() {
+    return consolidationRequestContractAddress;
+  }
+}
diff --git a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/transaction/TransactionSimulator.java b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/transaction/TransactionSimulator.java
index 070c1dcf771..29459ba3da4 100644
--- a/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/transaction/TransactionSimulator.java
+++ b/ethereum/core/src/main/java/org/hyperledger/besu/ethereum/transaction/TransactionSimulator.java
@@ -236,7 +236,9 @@ public Optional<TransactionSimulatorResult> processWithWorldUpdater(
             : blockHeaderToProcess.getGasLimit();
     if (rpcGasCap > 0) {
       gasLimit = rpcGasCap;
-      LOG.info("Capping gasLimit to " + rpcGasCap);
+      LOG.trace(
+          "Gas limit capped at {} for transaction simulation due to provided RPC gas cap.",
+          rpcGasCap);
     }
     final Wei value = callParams.getValue() != null ? callParams.getValue() : Wei.ZERO;
     final Bytes payload = callParams.getPayload() != null ? callParams.getPayload() : Bytes.EMPTY;
diff --git a/ethereum/core/src/test/java/org/hyperledger/besu/ethereum/mainnet/PragueRequestsValidatorTest.java b/ethereum/core/src/test/java/org/hyperledger/besu/ethereum/mainnet/PragueRequestsValidatorTest.java
index 6c325b70a86..6158ba44c37 100644
--- a/ethereum/core/src/test/java/org/hyperledger/besu/ethereum/mainnet/PragueRequestsValidatorTest.java
+++ b/ethereum/core/src/test/java/org/hyperledger/besu/ethereum/mainnet/PragueRequestsValidatorTest.java
@@ -17,8 +17,10 @@
 import static java.util.Collections.emptyList;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.hyperledger.besu.ethereum.mainnet.HeaderValidationMode.NONE;
+import static org.hyperledger.besu.ethereum.mainnet.requests.ConsolidationRequestProcessor.CONSOLIDATION_REQUEST_CONTRACT_ADDRESS;
 import static org.hyperledger.besu.ethereum.mainnet.requests.DepositRequestProcessor.DEFAULT_DEPOSIT_CONTRACT_ADDRESS;
 import static org.hyperledger.besu.ethereum.mainnet.requests.MainnetRequestsValidator.pragueRequestsValidator;
+import static org.hyperledger.besu.ethereum.mainnet.requests.WithdrawalRequestProcessor.DEFAULT_WITHDRAWAL_REQUEST_CONTRACT_ADDRESS;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.lenient;
 
@@ -31,6 +33,7 @@
 import org.hyperledger.besu.ethereum.core.BlockchainSetupUtil;
 import org.hyperledger.besu.ethereum.core.Request;
 import org.hyperledger.besu.ethereum.core.WithdrawalRequest;
+import org.hyperledger.besu.ethereum.mainnet.requests.RequestContractAddresses;
 import org.hyperledger.besu.ethereum.mainnet.requests.RequestsValidatorCoordinator;
 import org.hyperledger.besu.evm.log.LogsBloomFilter;
 
@@ -52,9 +55,13 @@ class PragueRequestsValidatorTest {
   @Mock private ProtocolSchedule protocolSchedule;
   @Mock private ProtocolSpec protocolSpec;
   @Mock private WithdrawalsValidator withdrawalsValidator;
+  private final RequestContractAddresses requestContractAddresses =
+      new RequestContractAddresses(
+          DEFAULT_WITHDRAWAL_REQUEST_CONTRACT_ADDRESS,
+          DEFAULT_DEPOSIT_CONTRACT_ADDRESS,
+          CONSOLIDATION_REQUEST_CONTRACT_ADDRESS);
 
-  RequestsValidatorCoordinator requestValidator =
-      pragueRequestsValidator(DEFAULT_DEPOSIT_CONTRACT_ADDRESS);
+  RequestsValidatorCoordinator requestValidator = pragueRequestsValidator(requestContractAddresses);
 
   @BeforeEach
   public void setUp() {
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java
index d46da85dc48..d72f76c213c 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/SynchronizerConfiguration.java
@@ -85,6 +85,7 @@ public class SynchronizerConfiguration {
   private final int maxTrailingPeers;
   private final long worldStateMinMillisBeforeStalling;
   private final long propagationManagerGetBlockTimeoutMillis;
+  private final boolean isPeerTaskSystemEnabled;
 
   private SynchronizerConfiguration(
       final int syncPivotDistance,
@@ -108,7 +109,8 @@ private SynchronizerConfiguration(
       final int computationParallelism,
       final int maxTrailingPeers,
       final long propagationManagerGetBlockTimeoutMillis,
-      final boolean checkpointPostMergeEnabled) {
+      final boolean checkpointPostMergeEnabled,
+      final boolean isPeerTaskSystemEnabled) {
     this.syncPivotDistance = syncPivotDistance;
     this.fastSyncFullValidationRate = fastSyncFullValidationRate;
     this.syncMinimumPeerCount = syncMinimumPeerCount;
@@ -131,6 +133,7 @@ private SynchronizerConfiguration(
     this.maxTrailingPeers = maxTrailingPeers;
     this.propagationManagerGetBlockTimeoutMillis = propagationManagerGetBlockTimeoutMillis;
     this.checkpointPostMergeEnabled = checkpointPostMergeEnabled;
+    this.isPeerTaskSystemEnabled = isPeerTaskSystemEnabled;
   }
 
   public static Builder builder() {
@@ -256,6 +259,10 @@ public long getPropagationManagerGetBlockTimeoutMillis() {
     return propagationManagerGetBlockTimeoutMillis;
   }
 
+  public boolean isPeerTaskSystemEnabled() {
+    return isPeerTaskSystemEnabled;
+  }
+
   public static class Builder {
     private SyncMode syncMode = SyncMode.FULL;
     private int syncMinimumPeerCount = DEFAULT_SYNC_MINIMUM_PEERS;
@@ -280,6 +287,7 @@ public static class Builder {
         DEFAULT_WORLD_STATE_MAX_REQUESTS_WITHOUT_PROGRESS;
     private long worldStateMinMillisBeforeStalling = DEFAULT_WORLD_STATE_MIN_MILLIS_BEFORE_STALLING;
     private int worldStateTaskCacheSize = DEFAULT_WORLD_STATE_TASK_CACHE_SIZE;
+    private boolean isPeerTaskSystemEnabled = false;
 
     private long propagationManagerGetBlockTimeoutMillis =
         DEFAULT_PROPAGATION_MANAGER_GET_BLOCK_TIMEOUT_MILLIS;
@@ -406,6 +414,11 @@ public Builder checkpointPostMergeEnabled(final boolean checkpointPostMergeEnabl
       return this;
     }
 
+    public Builder isPeerTaskSystemEnabled(final boolean isPeerTaskSystemEnabled) {
+      this.isPeerTaskSystemEnabled = isPeerTaskSystemEnabled;
+      return this;
+    }
+
     public SynchronizerConfiguration build() {
       return new SynchronizerConfiguration(
           syncPivotDistance,
@@ -429,7 +442,8 @@ public SynchronizerConfiguration build() {
           computationParallelism,
           maxTrailingPeers,
           propagationManagerGetBlockTimeoutMillis,
-          checkpointPostMergeEnabled);
+          checkpointPostMergeEnabled,
+          isPeerTaskSystemEnabled);
     }
   }
 }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/tasks/PersistBlockTask.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/tasks/PersistBlockTask.java
index 85a478c9aaf..6333f0d1913 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/tasks/PersistBlockTask.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/tasks/PersistBlockTask.java
@@ -218,7 +218,8 @@ protected void cleanup() {
       case IMPORTED:
         LOG.info(
             String.format(
-                "Imported #%,d / %d tx / %d om / %,d (%01.1f%%) gas / (%s) in %01.3fs. Peers: %d",
+                "Imported %s #%,d / %d tx / %d om / %,d (%01.1f%%) gas / (%s) in %01.3fs. Peers: %d",
+                block.getBody().getTransactions().size() == 0 ? "empty block" : "block",
                 block.getHeader().getNumber(),
                 block.getBody().getTransactions().size(),
                 block.getBody().getOmmers().size(),
diff --git a/testutil/src/main/resources/log4j2.xml b/testutil/src/main/resources/log4j2-test.xml
similarity index 100%
rename from testutil/src/main/resources/log4j2.xml
rename to testutil/src/main/resources/log4j2-test.xml