diff --git a/CHANGELOG.md b/CHANGELOG.md
index e95e9d9c239..dce18aaeaeb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,10 +12,10 @@
 * Ibft2 will discard any received messages targeting a chain height <= current head - this resolves some corner cases in system correctness directly following block import. [#1575](https://github.com/hyperledger/besu/pull/1575)
 * EvmTool now throws `UnsupportedForkException` when there is an unknown fork and is YOLOv2 compatible [\#1584](https://github.com/hyperledger/besu/pull/1584)
 * `eth_newFilter` now supports `blockHash` parameter as per the spec [\#1548](https://github.com/hyperledger/besu/issues/1540). (`blockhash` is also still supported.)
+* Fixed an issue that caused loss of peers and desynchronization when eth65 was enabled [\#1601](https://github.com/hyperledger/besu/pull/1601)
 
 #### Previously identified known issues
 
-- [Eth/65 loses peers](KNOWN_ISSUES.md#eth65-loses-peers)
 - [Fast sync when running Besu on cloud providers](KNOWN_ISSUES.md#fast-sync-when-running-besu-on-cloud-providers)
 - [Privacy users with private transactions created using v1.3.4 or earlier](KNOWN_ISSUES.md#privacy-users-with-private-transactions-created-using-v134-or-earlier)
 
diff --git a/KNOWN_ISSUES.md b/KNOWN_ISSUES.md
index 659ffbd2f79..89c67789eda 100644
--- a/KNOWN_ISSUES.md
+++ b/KNOWN_ISSUES.md
@@ -5,14 +5,6 @@ in the current release are provided in the [Changelog](CHANGELOG.md).
 
 Known issues are open issues categorized as [Very High or High impact](https://wiki.hyperledger.org/display/BESU/Defect+Prioritisation+Policy). 
 
-## Eth/65 loses peers 
-
-From v1.4.4, `eth/65` is [disabled by default](https://github.com/hyperledger/besu/pull/741). 
-
-If enabled, peers will slowly drop off and eventually Besu will fall out of sync or stop syncing.
-
-A fix for this issue is being actively worked on. 
-
 ## Fast sync when running Besu on cloud providers  
 
 A known [RocksDB issue](https://github.com/facebook/rocksdb/issues/6435) causes fast sync to fail 
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 5d6c2851bc3..48fe97eb98a 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
@@ -156,7 +156,7 @@ public void startNode(final BesuNode node) {
             .privacyParameters(node.getPrivacyParameters())
             .nodeKey(new NodeKey(new KeyPairSecurityModule(KeyPairUtil.loadKeyPair(dataDir))))
             .metricsSystem(metricsSystem)
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .ethProtocolConfiguration(EthProtocolConfiguration.defaultConfig())
             .clock(Clock.systemUTC())
             .isRevertReasonEnabled(node.isRevertReasonEnabled())
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 6c813828b96..f3460f8d089 100644
--- a/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java
+++ b/besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java
@@ -146,6 +146,7 @@
 import org.hyperledger.besu.util.NetworkUtility;
 import org.hyperledger.besu.util.PermissioningConfigurationValidator;
 import org.hyperledger.besu.util.number.Fraction;
+import org.hyperledger.besu.util.number.Percentage;
 import org.hyperledger.besu.util.number.PositiveNumber;
 
 import java.io.File;
@@ -2077,7 +2078,7 @@ private TransactionPoolConfiguration buildTransactionPoolConfiguration() {
         .txPoolMaxSize(txPoolMaxSize)
         .pooledTransactionHashesSize(pooledTransactionHashesSize)
         .pendingTxRetentionPeriod(pendingTxRetentionPeriod)
-        .priceBump(priceBump)
+        .priceBump(Percentage.fromInt(priceBump))
         .txFeeCap(txFeeCap)
         .build();
   }
diff --git a/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/TransactionPoolOptions.java b/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/TransactionPoolOptions.java
index bcb33d9a945..cbfa263af43 100644
--- a/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/TransactionPoolOptions.java
+++ b/besu/src/main/java/org/hyperledger/besu/cli/options/unstable/TransactionPoolOptions.java
@@ -16,17 +16,23 @@
 
 import org.hyperledger.besu.cli.options.CLIOptions;
 import org.hyperledger.besu.cli.options.OptionParser;
+import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 
 import picocli.CommandLine;
 
-public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfiguration.Builder> {
+public class TransactionPoolOptions
+    implements CLIOptions<ImmutableTransactionPoolConfiguration.Builder> {
   private static final String TX_MESSAGE_KEEP_ALIVE_SEC_FLAG =
       "--Xincoming-tx-messages-keep-alive-seconds";
 
+  private static final String ETH65_TX_ANNOUNCED_BUFFERING_PERIOD_FLAG =
+      "--Xeth65-tx-announced-buffering-period-milliseconds";
+
   @CommandLine.Option(
       names = {TX_MESSAGE_KEEP_ALIVE_SEC_FLAG},
       paramLabel = "<INTEGER>",
@@ -37,6 +43,16 @@ public class TransactionPoolOptions implements CLIOptions<TransactionPoolConfigu
   private Integer txMessageKeepAliveSeconds =
       TransactionPoolConfiguration.DEFAULT_TX_MSG_KEEP_ALIVE;
 
+  @CommandLine.Option(
+      names = {ETH65_TX_ANNOUNCED_BUFFERING_PERIOD_FLAG},
+      paramLabel = "<LONG>",
+      hidden = true,
+      description =
+          "The period for which the announced transactions remain in the buffer before being requested from the peers in milliseconds (default: ${DEFAULT-VALUE})",
+      arity = "1")
+  private long eth65TrxAnnouncedBufferingPeriod =
+      TransactionPoolConfiguration.ETH65_TRX_ANNOUNCED_BUFFERING_PERIOD.toMillis();
+
   private TransactionPoolOptions() {}
 
   public static TransactionPoolOptions create() {
@@ -46,18 +62,24 @@ public static TransactionPoolOptions create() {
   public static TransactionPoolOptions fromConfig(final TransactionPoolConfiguration config) {
     final TransactionPoolOptions options = TransactionPoolOptions.create();
     options.txMessageKeepAliveSeconds = config.getTxMessageKeepAliveSeconds();
+    options.eth65TrxAnnouncedBufferingPeriod =
+        config.getEth65TrxAnnouncedBufferingPeriod().toMillis();
     return options;
   }
 
   @Override
-  public TransactionPoolConfiguration.Builder toDomainObject() {
-    return TransactionPoolConfiguration.builder()
-        .txMessageKeepAliveSeconds(txMessageKeepAliveSeconds);
+  public ImmutableTransactionPoolConfiguration.Builder toDomainObject() {
+    return ImmutableTransactionPoolConfiguration.builder()
+        .txMessageKeepAliveSeconds(txMessageKeepAliveSeconds)
+        .eth65TrxAnnouncedBufferingPeriod(Duration.ofMillis(eth65TrxAnnouncedBufferingPeriod));
   }
 
   @Override
   public List<String> getCLIOptions() {
     return Arrays.asList(
-        TX_MESSAGE_KEEP_ALIVE_SEC_FLAG, OptionParser.format(txMessageKeepAliveSeconds));
+        TX_MESSAGE_KEEP_ALIVE_SEC_FLAG,
+        OptionParser.format(txMessageKeepAliveSeconds),
+        ETH65_TX_ANNOUNCED_BUFFERING_PERIOD_FLAG,
+        OptionParser.format(eth65TrxAnnouncedBufferingPeriod));
   }
 }
diff --git a/besu/src/test/java/org/hyperledger/besu/PrivacyReorgTest.java b/besu/src/test/java/org/hyperledger/besu/PrivacyReorgTest.java
index 576d211ec33..8314695db32 100644
--- a/besu/src/test/java/org/hyperledger/besu/PrivacyReorgTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/PrivacyReorgTest.java
@@ -167,7 +167,7 @@ public void setUp() throws IOException {
             .dataDirectory(dataDir)
             .clock(TestClock.fixed())
             .privacyParameters(privacyParameters)
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .gasLimitCalculator(GasLimitCalculator.constant())
             .build();
   }
diff --git a/besu/src/test/java/org/hyperledger/besu/PrivacyTest.java b/besu/src/test/java/org/hyperledger/besu/PrivacyTest.java
index 2721231c54d..be171239de3 100644
--- a/besu/src/test/java/org/hyperledger/besu/PrivacyTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/PrivacyTest.java
@@ -113,7 +113,7 @@ private BesuController setUpControllerWithPrivacyEnabled(final boolean onChainEn
         .dataDirectory(dataDir)
         .clock(TestClock.fixed())
         .privacyParameters(privacyParameters)
-        .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+        .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
         .gasLimitCalculator(GasLimitCalculator.constant())
         .build();
   }
diff --git a/besu/src/test/java/org/hyperledger/besu/RunnerTest.java b/besu/src/test/java/org/hyperledger/besu/RunnerTest.java
index 0d8aa45efc6..3616efe420b 100644
--- a/besu/src/test/java/org/hyperledger/besu/RunnerTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/RunnerTest.java
@@ -164,7 +164,7 @@ private void syncFromGenesis(final SyncMode mode, final GenesisConfigFile genesi
             .metricsSystem(noOpMetricsSystem)
             .privacyParameters(PrivacyParameters.DEFAULT)
             .clock(TestClock.fixed())
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .storageProvider(createKeyValueStorageProvider(dataDirAhead, dbAhead))
             .gasLimitCalculator(GasLimitCalculator.constant())
             .build()) {
@@ -184,7 +184,7 @@ private void syncFromGenesis(final SyncMode mode, final GenesisConfigFile genesi
             .metricsSystem(noOpMetricsSystem)
             .privacyParameters(PrivacyParameters.DEFAULT)
             .clock(TestClock.fixed())
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .storageProvider(createKeyValueStorageProvider(dataDirAhead, dbAhead))
             .gasLimitCalculator(GasLimitCalculator.constant())
             .build();
@@ -248,7 +248,7 @@ private void syncFromGenesis(final SyncMode mode, final GenesisConfigFile genesi
               .metricsSystem(noOpMetricsSystem)
               .privacyParameters(PrivacyParameters.DEFAULT)
               .clock(TestClock.fixed())
-              .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+              .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
               .gasLimitCalculator(GasLimitCalculator.constant())
               .build();
       final EnodeURL enode = runnerAhead.getLocalEnode().get();
diff --git a/besu/src/test/java/org/hyperledger/besu/chainexport/RlpBlockExporterTest.java b/besu/src/test/java/org/hyperledger/besu/chainexport/RlpBlockExporterTest.java
index 2f4d1689f6c..cd97ad53cb7 100644
--- a/besu/src/test/java/org/hyperledger/besu/chainexport/RlpBlockExporterTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/chainexport/RlpBlockExporterTest.java
@@ -90,7 +90,7 @@ private static BesuController createController() throws IOException {
         .privacyParameters(PrivacyParameters.DEFAULT)
         .dataDirectory(dataDir)
         .clock(TestClock.fixed())
-        .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+        .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
         .gasLimitCalculator(GasLimitCalculator.constant())
         .build();
   }
diff --git a/besu/src/test/java/org/hyperledger/besu/chainimport/JsonBlockImporterTest.java b/besu/src/test/java/org/hyperledger/besu/chainimport/JsonBlockImporterTest.java
index 8840476f982..d22d31dfc03 100644
--- a/besu/src/test/java/org/hyperledger/besu/chainimport/JsonBlockImporterTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/chainimport/JsonBlockImporterTest.java
@@ -427,7 +427,7 @@ protected BesuController createController(final GenesisConfigFile genesisConfigF
         .privacyParameters(PrivacyParameters.DEFAULT)
         .dataDirectory(dataDir)
         .clock(TestClock.fixed())
-        .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+        .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
         .gasLimitCalculator(GasLimitCalculator.constant())
         .build();
   }
diff --git a/besu/src/test/java/org/hyperledger/besu/chainimport/RlpBlockImporterTest.java b/besu/src/test/java/org/hyperledger/besu/chainimport/RlpBlockImporterTest.java
index dd924e1cd66..13bd58428da 100644
--- a/besu/src/test/java/org/hyperledger/besu/chainimport/RlpBlockImporterTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/chainimport/RlpBlockImporterTest.java
@@ -70,7 +70,7 @@ public void blockImport() throws IOException {
             .privacyParameters(PrivacyParameters.DEFAULT)
             .dataDirectory(dataDir)
             .clock(TestClock.fixed())
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .gasLimitCalculator(GasLimitCalculator.constant())
             .build();
     final RlpBlockImporter.ImportResult result =
@@ -98,7 +98,7 @@ public void blockImportRejectsBadPow() throws IOException {
             .privacyParameters(PrivacyParameters.DEFAULT)
             .dataDirectory(dataDir)
             .clock(TestClock.fixed())
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .gasLimitCalculator(GasLimitCalculator.constant())
             .build();
 
@@ -126,7 +126,7 @@ public void blockImportCanSkipPow() throws IOException {
             .privacyParameters(PrivacyParameters.DEFAULT)
             .dataDirectory(dataDir)
             .clock(TestClock.fixed())
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .gasLimitCalculator(GasLimitCalculator.constant())
             .build();
 
@@ -166,7 +166,7 @@ public void ibftImport() throws IOException {
             .privacyParameters(PrivacyParameters.DEFAULT)
             .dataDirectory(dataDir)
             .clock(TestClock.fixed())
-            .transactionPoolConfiguration(TransactionPoolConfiguration.builder().build())
+            .transactionPoolConfiguration(TransactionPoolConfiguration.DEFAULT)
             .gasLimitCalculator(GasLimitCalculator.constant())
             .build();
     final RlpBlockImporter.ImportResult result =
diff --git a/besu/src/test/java/org/hyperledger/besu/cli/BesuCommandTest.java b/besu/src/test/java/org/hyperledger/besu/cli/BesuCommandTest.java
index b4584766939..6dd478c38ee 100644
--- a/besu/src/test/java/org/hyperledger/besu/cli/BesuCommandTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/cli/BesuCommandTest.java
@@ -3411,6 +3411,18 @@ public void txMessageKeepAliveSecondsWithInvalidInputShouldFail() {
             "Invalid value for option '--Xincoming-tx-messages-keep-alive-seconds': 'acbd' is not an int");
   }
 
+  @Test
+  public void eth65TrxAnnouncedBufferingPeriodWithInvalidInputShouldFail() {
+    parseCommand("--Xeth65-tx-announced-buffering-period-milliseconds", "acbd");
+
+    Mockito.verifyZeroInteractions(mockRunnerBuilder);
+
+    assertThat(commandOutput.toString()).isEmpty();
+    assertThat(commandErrorOutput.toString())
+        .contains(
+            "Invalid value for option '--Xeth65-tx-announced-buffering-period-milliseconds': 'acbd' is not a long");
+  }
+
   @Test
   public void tomlThatHasInvalidOptions() throws IOException {
     final URL configFile = this.getClass().getResource("/complete_config.toml");
diff --git a/besu/src/test/java/org/hyperledger/besu/cli/options/TransactionPoolOptionsTest.java b/besu/src/test/java/org/hyperledger/besu/cli/options/TransactionPoolOptionsTest.java
index 15d074ffd46..c9313d39889 100644
--- a/besu/src/test/java/org/hyperledger/besu/cli/options/TransactionPoolOptionsTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/cli/options/TransactionPoolOptionsTest.java
@@ -17,12 +17,16 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import org.hyperledger.besu.cli.options.unstable.TransactionPoolOptions;
+import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
 
+import java.time.Duration;
+
 import org.junit.Test;
 
 public class TransactionPoolOptionsTest
-    extends AbstractCLIOptionsTest<TransactionPoolConfiguration.Builder, TransactionPoolOptions> {
+    extends AbstractCLIOptionsTest<
+        ImmutableTransactionPoolConfiguration.Builder, TransactionPoolOptions> {
 
   @Test
   public void txMessageKeepAliveSeconds() {
@@ -40,20 +44,44 @@ public void txMessageKeepAliveSeconds() {
     assertThat(commandErrorOutput.toString()).isEmpty();
   }
 
+  @Test
+  public void eth65TrxAnnouncedBufferingPeriod() {
+    final long eth65TrxAnnouncedBufferingPeriod = 999;
+    final TestBesuCommand cmd =
+        parseCommand(
+            "--Xeth65-tx-announced-buffering-period-milliseconds",
+            String.valueOf(eth65TrxAnnouncedBufferingPeriod));
+
+    final TransactionPoolOptions options = getOptionsFromBesuCommand(cmd);
+    final TransactionPoolConfiguration config = options.toDomainObject().build();
+    assertThat(config.getEth65TrxAnnouncedBufferingPeriod())
+        .hasMillis(eth65TrxAnnouncedBufferingPeriod);
+
+    assertThat(commandOutput.toString()).isEmpty();
+    assertThat(commandErrorOutput.toString()).isEmpty();
+  }
+
   @Override
-  TransactionPoolConfiguration.Builder createDefaultDomainObject() {
-    return TransactionPoolConfiguration.builder();
+  ImmutableTransactionPoolConfiguration.Builder createDefaultDomainObject() {
+    final ImmutableTransactionPoolConfiguration defaultValue =
+        ImmutableTransactionPoolConfiguration.builder().build();
+    return ImmutableTransactionPoolConfiguration.builder()
+        .txMessageKeepAliveSeconds(defaultValue.getTxMessageKeepAliveSeconds())
+        .eth65TrxAnnouncedBufferingPeriod(defaultValue.getEth65TrxAnnouncedBufferingPeriod());
   }
 
   @Override
-  TransactionPoolConfiguration.Builder createCustomizedDomainObject() {
-    return TransactionPoolConfiguration.builder()
-        .txMessageKeepAliveSeconds(TransactionPoolConfiguration.DEFAULT_TX_MSG_KEEP_ALIVE + 1);
+  ImmutableTransactionPoolConfiguration.Builder createCustomizedDomainObject() {
+    return ImmutableTransactionPoolConfiguration.builder()
+        .txMessageKeepAliveSeconds(TransactionPoolConfiguration.DEFAULT_TX_MSG_KEEP_ALIVE + 1)
+        .eth65TrxAnnouncedBufferingPeriod(
+            TransactionPoolConfiguration.ETH65_TRX_ANNOUNCED_BUFFERING_PERIOD.plus(
+                Duration.ofMillis(100)));
   }
 
   @Override
   TransactionPoolOptions optionsFromDomainObject(
-      final TransactionPoolConfiguration.Builder domainObject) {
+      final ImmutableTransactionPoolConfiguration.Builder domainObject) {
     return TransactionPoolOptions.fromConfig(domainObject.build());
   }
 
diff --git a/besu/src/test/java/org/hyperledger/besu/services/BesuEventsImplTest.java b/besu/src/test/java/org/hyperledger/besu/services/BesuEventsImplTest.java
index 1592564f302..4a06d63e66f 100644
--- a/besu/src/test/java/org/hyperledger/besu/services/BesuEventsImplTest.java
+++ b/besu/src/test/java/org/hyperledger/besu/services/BesuEventsImplTest.java
@@ -40,6 +40,7 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.sync.BlockBroadcaster;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
+import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolFactory;
@@ -129,7 +130,7 @@ public void setUp() {
     blockBroadcaster = new BlockBroadcaster(mockEthContext);
     syncState = new SyncState(blockchain, mockEthPeers);
     TransactionPoolConfiguration txPoolConfig =
-        TransactionPoolConfiguration.builder().txPoolMaxSize(1).build();
+        ImmutableTransactionPoolConfiguration.builder().txPoolMaxSize(1).build();
 
     transactionPool =
         TransactionPoolFactory.createTransactionPool(
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/task/BufferedGetPooledTransactionsFromPeerFetcher.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/task/BufferedGetPooledTransactionsFromPeerFetcher.java
new file mode 100644
index 00000000000..8332ccaa88d
--- /dev/null
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/task/BufferedGetPooledTransactionsFromPeerFetcher.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * 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.eth.manager.task;
+
+import static org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration.MAX_PENDING_TRANSACTIONS;
+
+import org.hyperledger.besu.ethereum.core.Hash;
+import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionsMessageProcessor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+
+import com.google.common.collect.EvictingQueue;
+import com.google.common.collect.Queues;
+
+@SuppressWarnings("UnstableApiUsage")
+public class BufferedGetPooledTransactionsFromPeerFetcher {
+
+  private static final int MAX_HASHES = 256;
+
+  private final EthPeer peer;
+  private final PendingTransactionsMessageProcessor processor;
+  private final Queue<Hash> txAnnounces;
+
+  public BufferedGetPooledTransactionsFromPeerFetcher(
+      final EthPeer peer, final PendingTransactionsMessageProcessor processor) {
+    this.peer = peer;
+    this.processor = processor;
+    this.txAnnounces = Queues.synchronizedQueue(EvictingQueue.create(MAX_PENDING_TRANSACTIONS));
+  }
+
+  public void requestTransactions() {
+    for (List<Hash> txAnnounces = getTxAnnounces();
+        !txAnnounces.isEmpty();
+        txAnnounces = getTxAnnounces()) {
+      final GetPooledTransactionsFromPeerTask task =
+          GetPooledTransactionsFromPeerTask.forHashes(
+              processor.getEthContext(), txAnnounces, processor.getMetricsSystem());
+      task.assignPeer(peer);
+      processor
+          .getEthContext()
+          .getScheduler()
+          .scheduleSyncWorkerTask(task)
+          .thenAccept(
+              result -> processor.getTransactionPool().addRemoteTransactions(result.getResult()));
+    }
+  }
+
+  public void addHash(final Hash hash) {
+    txAnnounces.add(hash);
+  }
+
+  private List<Hash> getTxAnnounces() {
+    List<Hash> retrieved = new ArrayList<>();
+    while (retrieved.size() < MAX_HASHES && !txAnnounces.isEmpty()) {
+      final Hash txAnnounce = txAnnounces.poll();
+      if (processor.getTransactionPool().getTransactionByHash(txAnnounce).isEmpty()) {
+        retrieved.add(txAnnounce);
+      }
+    }
+    return retrieved;
+  }
+}
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessor.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessor.java
index 9533f0d8ac4..3c6bde13f99 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessor.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessor.java
@@ -18,10 +18,9 @@
 import static org.apache.logging.log4j.LogManager.getLogger;
 
 import org.hyperledger.besu.ethereum.core.Hash;
-import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
-import org.hyperledger.besu.ethereum.eth.manager.task.GetPooledTransactionsFromPeerTask;
+import org.hyperledger.besu.ethereum.eth.manager.task.BufferedGetPooledTransactionsFromPeerFetcher;
 import org.hyperledger.besu.ethereum.eth.messages.NewPooledTransactionHashesMessage;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage.DisconnectReason;
@@ -32,34 +31,39 @@
 
 import java.time.Duration;
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.logging.log4j.Logger;
 
-class PendingTransactionsMessageProcessor {
+public class PendingTransactionsMessageProcessor {
 
-  private static final int MAX_HASHES = 256;
   private static final int SKIPPED_MESSAGES_LOGGING_THRESHOLD = 1000;
   private static final long SYNC_TOLERANCE = 100L;
 
   private static final Logger LOG = getLogger();
+
+  private final ConcurrentHashMap<EthPeer, BufferedGetPooledTransactionsFromPeerFetcher>
+      scheduledTasks;
+
   private final PeerPendingTransactionTracker transactionTracker;
   private final Counter totalSkippedTransactionsMessageCounter;
   private final TransactionPool transactionPool;
+  private final TransactionPoolConfiguration transactionPoolConfiguration;
   private final EthContext ethContext;
   private final MetricsSystem metricsSystem;
   private final SyncState syncState;
 
-  PendingTransactionsMessageProcessor(
+  public PendingTransactionsMessageProcessor(
       final PeerPendingTransactionTracker transactionTracker,
       final TransactionPool transactionPool,
+      final TransactionPoolConfiguration transactionPoolConfiguration,
       final Counter metricsCounter,
       final EthContext ethContext,
       final MetricsSystem metricsSystem,
       final SyncState syncState) {
     this.transactionTracker = transactionTracker;
     this.transactionPool = transactionPool;
+    this.transactionPoolConfiguration = transactionPoolConfiguration;
     this.ethContext = ethContext;
     this.metricsSystem = metricsSystem;
     this.syncState = syncState;
@@ -71,6 +75,7 @@ class PendingTransactionsMessageProcessor {
                     "{} expired transaction messages have been skipped.",
                     SKIPPED_MESSAGES_LOGGING_THRESHOLD),
             SKIPPED_MESSAGES_LOGGING_THRESHOLD);
+    this.scheduledTasks = new ConcurrentHashMap<>();
   }
 
   void processNewPooledTransactionHashesMessage(
@@ -86,37 +91,33 @@ void processNewPooledTransactionHashesMessage(
     }
   }
 
+  @SuppressWarnings("UnstableApiUsage")
   private void processNewPooledTransactionHashesMessage(
       final EthPeer peer, final NewPooledTransactionHashesMessage transactionsMessage) {
     try {
       LOG.trace("Received pooled transaction hashes message from {}", peer);
 
-      final List<Hash> pendingHashes = transactionsMessage.pendingTransactions();
-      transactionTracker.markTransactionsHashesAsSeen(peer, pendingHashes);
+      transactionTracker.markTransactionsHashesAsSeen(
+          peer, transactionsMessage.pendingTransactions());
       if (syncState.isInSync(SYNC_TOLERANCE)) {
-        final List<Hash> toRequest = new ArrayList<>();
-        for (final Hash hash : pendingHashes) {
-          if (transactionPool.addTransactionHash(hash)) {
-            toRequest.add(hash);
+        final BufferedGetPooledTransactionsFromPeerFetcher bufferedTask =
+            scheduledTasks.computeIfAbsent(
+                peer,
+                ethPeer -> {
+                  ethContext
+                      .getScheduler()
+                      .scheduleFutureTask(
+                          new FetcherCreatorTask(peer),
+                          transactionPoolConfiguration.getEth65TrxAnnouncedBufferingPeriod());
+                  return new BufferedGetPooledTransactionsFromPeerFetcher(peer, this);
+                });
+
+        for (final Hash hash : transactionsMessage.pendingTransactions()) {
+          if (transactionPool.getTransactionByHash(hash).isEmpty()
+              && transactionPool.addTransactionHash(hash)) {
+            bufferedTask.addHash(hash);
           }
         }
-        while (!toRequest.isEmpty()) {
-          final List<Hash> messageHashes =
-              toRequest.subList(0, Math.min(toRequest.size(), MAX_HASHES));
-          final GetPooledTransactionsFromPeerTask task =
-              GetPooledTransactionsFromPeerTask.forHashes(ethContext, messageHashes, metricsSystem);
-          task.assignPeer(peer);
-          ethContext
-              .getScheduler()
-              .scheduleSyncWorkerTask(task)
-              .thenAccept(
-                  result -> {
-                    final List<Transaction> txs = result.getResult();
-                    transactionPool.addRemoteTransactions(txs);
-                  });
-
-          toRequest.removeAll(messageHashes);
-        }
       }
     } catch (final RLPException ex) {
       if (peer != null) {
@@ -126,4 +127,34 @@ private void processNewPooledTransactionHashesMessage(
       }
     }
   }
+
+  public TransactionPool getTransactionPool() {
+    return transactionPool;
+  }
+
+  public EthContext getEthContext() {
+    return ethContext;
+  }
+
+  public MetricsSystem getMetricsSystem() {
+    return metricsSystem;
+  }
+
+  public class FetcherCreatorTask implements Runnable {
+    final EthPeer peer;
+
+    public FetcherCreatorTask(final EthPeer peer) {
+      this.peer = peer;
+    }
+
+    @Override
+    public void run() {
+      if (peer != null) {
+        final BufferedGetPooledTransactionsFromPeerFetcher fetcher = scheduledTasks.remove(peer);
+        if (!peer.isDisconnected()) {
+          fetcher.requestTransactions();
+        }
+      }
+    }
+  }
 }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolConfiguration.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolConfiguration.java
index 74268c510ca..2401278c3d6 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolConfiguration.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolConfiguration.java
@@ -17,157 +17,55 @@
 import org.hyperledger.besu.ethereum.core.Wei;
 import org.hyperledger.besu.util.number.Percentage;
 
-import java.util.Objects;
-
-public class TransactionPoolConfiguration {
-  public static final int DEFAULT_TX_MSG_KEEP_ALIVE = 60;
-  public static final int MAX_PENDING_TRANSACTIONS = 4096;
-  public static final int MAX_PENDING_TRANSACTIONS_HASHES = 4096;
-  public static final int DEFAULT_TX_RETENTION_HOURS = 13;
-  public static final Percentage DEFAULT_PRICE_BUMP = Percentage.fromInt(10);
-  public static final Wei DEFAULT_RPC_TX_FEE_CAP = Wei.fromEth(1);
-  public static final TransactionPoolConfiguration DEFAULT =
-      TransactionPoolConfiguration.builder().build();
-
-  private final int txPoolMaxSize;
-  private final int pooledTransactionHashesSize;
-  private final int pendingTxRetentionPeriod;
-  private final int txMessageKeepAliveSeconds;
-  private final Percentage priceBump;
-
-  private final Wei txFeeCap;
-
-  public TransactionPoolConfiguration(
-      final int txPoolMaxSize,
-      final int pooledTransactionHashesSize,
-      final int pendingTxRetentionPeriod,
-      final int txMessageKeepAliveSeconds,
-      final Percentage priceBump,
-      final Wei txFeeCap) {
-    this.txPoolMaxSize = txPoolMaxSize;
-    this.pooledTransactionHashesSize = pooledTransactionHashesSize;
-    this.pendingTxRetentionPeriod = pendingTxRetentionPeriod;
-    this.txMessageKeepAliveSeconds = txMessageKeepAliveSeconds;
-    this.priceBump = priceBump;
-    this.txFeeCap = txFeeCap;
-  }
-
-  public int getTxPoolMaxSize() {
-    return txPoolMaxSize;
-  }
-
-  public int getPooledTransactionHashesSize() {
-    return pooledTransactionHashesSize;
-  }
-
-  public int getPendingTxRetentionPeriod() {
-    return pendingTxRetentionPeriod;
-  }
-
-  public int getTxMessageKeepAliveSeconds() {
-    return txMessageKeepAliveSeconds;
-  }
-
-  public Percentage getPriceBump() {
-    return priceBump;
+import java.time.Duration;
+
+import org.immutables.value.Value;
+
+@Value.Immutable
+@Value.Style(allParameters = true)
+public interface TransactionPoolConfiguration {
+  int DEFAULT_TX_MSG_KEEP_ALIVE = 60;
+  int MAX_PENDING_TRANSACTIONS = 4096;
+  int MAX_PENDING_TRANSACTIONS_HASHES = 4096;
+  int DEFAULT_TX_RETENTION_HOURS = 13;
+  Percentage DEFAULT_PRICE_BUMP = Percentage.fromInt(10);
+  Wei DEFAULT_RPC_TX_FEE_CAP = Wei.fromEth(1);
+  Duration ETH65_TRX_ANNOUNCED_BUFFERING_PERIOD = Duration.ofMillis(500);
+
+  TransactionPoolConfiguration DEFAULT = ImmutableTransactionPoolConfiguration.builder().build();
+
+  @Value.Default
+  default int getTxPoolMaxSize() {
+    return MAX_PENDING_TRANSACTIONS;
   }
 
-  public Wei getTxFeeCap() {
-    return txFeeCap;
+  @Value.Default
+  default int getPooledTransactionHashesSize() {
+    return MAX_PENDING_TRANSACTIONS_HASHES;
   }
 
-  @Override
-  public boolean equals(final Object o) {
-    if (this == o) {
-      return true;
-    }
-    if (o == null || getClass() != o.getClass()) {
-      return false;
-    }
-    final TransactionPoolConfiguration that = (TransactionPoolConfiguration) o;
-    return txPoolMaxSize == that.txPoolMaxSize
-        && Objects.equals(pendingTxRetentionPeriod, that.pendingTxRetentionPeriod)
-        && Objects.equals(txMessageKeepAliveSeconds, that.txMessageKeepAliveSeconds)
-        && Objects.equals(priceBump, that.priceBump)
-        && Objects.equals(txFeeCap, that.txFeeCap);
+  @Value.Default
+  default int getPendingTxRetentionPeriod() {
+    return DEFAULT_TX_RETENTION_HOURS;
   }
 
-  @Override
-  public int hashCode() {
-    return Objects.hash(
-        txPoolMaxSize, pendingTxRetentionPeriod, txMessageKeepAliveSeconds, priceBump, txFeeCap);
+  @Value.Default
+  default int getTxMessageKeepAliveSeconds() {
+    return DEFAULT_TX_MSG_KEEP_ALIVE;
   }
 
-  @Override
-  public String toString() {
-    return "TransactionPoolConfiguration{"
-        + "txPoolMaxSize="
-        + txPoolMaxSize
-        + ", pendingTxRetentionPeriod="
-        + pendingTxRetentionPeriod
-        + ", txMessageKeepAliveSeconds="
-        + txMessageKeepAliveSeconds
-        + ", priceBump="
-        + priceBump
-        + ", txFeeCap="
-        + txFeeCap
-        + '}';
+  @Value.Default
+  default Percentage getPriceBump() {
+    return DEFAULT_PRICE_BUMP;
   }
 
-  public static Builder builder() {
-    return new Builder();
+  @Value.Default
+  default Duration getEth65TrxAnnouncedBufferingPeriod() {
+    return ETH65_TRX_ANNOUNCED_BUFFERING_PERIOD;
   }
 
-  public static class Builder {
-    private int txPoolMaxSize = MAX_PENDING_TRANSACTIONS;
-    private int pendingTxRetentionPeriod = DEFAULT_TX_RETENTION_HOURS;
-    private Integer txMessageKeepAliveSeconds = DEFAULT_TX_MSG_KEEP_ALIVE;
-    private int pooledTransactionHashesSize = MAX_PENDING_TRANSACTIONS_HASHES;
-    private Percentage priceBump = DEFAULT_PRICE_BUMP;
-    private Wei txFeeCap = DEFAULT_RPC_TX_FEE_CAP;
-
-    public Builder txPoolMaxSize(final int txPoolMaxSize) {
-      this.txPoolMaxSize = txPoolMaxSize;
-      return this;
-    }
-
-    public Builder pooledTransactionHashesSize(final int pooledTransactionHashesSize) {
-      this.pooledTransactionHashesSize = pooledTransactionHashesSize;
-      return this;
-    }
-
-    public Builder pendingTxRetentionPeriod(final int pendingTxRetentionPeriod) {
-      this.pendingTxRetentionPeriod = pendingTxRetentionPeriod;
-      return this;
-    }
-
-    public Builder txMessageKeepAliveSeconds(final int txMessageKeepAliveSeconds) {
-      this.txMessageKeepAliveSeconds = txMessageKeepAliveSeconds;
-      return this;
-    }
-
-    public Builder priceBump(final Percentage priceBump) {
-      this.priceBump = priceBump;
-      return this;
-    }
-
-    public Builder priceBump(final int priceBump) {
-      return priceBump(Percentage.fromInt(priceBump));
-    }
-
-    public Builder txFeeCap(final Wei txFeeCap) {
-      this.txFeeCap = txFeeCap;
-      return this;
-    }
-
-    public TransactionPoolConfiguration build() {
-      return new TransactionPoolConfiguration(
-          txPoolMaxSize,
-          pooledTransactionHashesSize,
-          pendingTxRetentionPeriod,
-          txMessageKeepAliveSeconds,
-          priceBump,
-          txFeeCap);
-    }
+  @Value.Default
+  default Wei getTxFeeCap() {
+    return DEFAULT_RPC_TX_FEE_CAP;
   }
 }
diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactory.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactory.java
index a4b50081371..b23248519bd 100644
--- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactory.java
+++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactory.java
@@ -146,6 +146,7 @@ static TransactionPool createTransactionPool(
               new PendingTransactionsMessageProcessor(
                   pendingTransactionTracker.get(),
                   transactionPool,
+                  transactionPoolConfiguration,
                   metricsSystem.createCounter(
                       BesuMetricCategory.TRANSACTION_POOL,
                       "pending_transactions_messages_skipped_total",
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManagerTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManagerTest.java
index fdfd5a571f0..ad3f415c28d 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManagerTest.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/EthProtocolManagerTest.java
@@ -987,7 +987,7 @@ public void transactionMessagesGoToTheCorrectExecutor() {
           metricsSystem,
           mock(SyncState.class),
           Wei.ZERO,
-          TransactionPoolConfiguration.builder().build(),
+          TransactionPoolConfiguration.DEFAULT,
           true,
           Optional.empty());
 
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/ethtaskutils/AbstractMessageTaskTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/ethtaskutils/AbstractMessageTaskTest.java
index 1442d53f581..333ae2c7c3d 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/ethtaskutils/AbstractMessageTaskTest.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/ethtaskutils/AbstractMessageTaskTest.java
@@ -99,7 +99,7 @@ public void setupTest() {
             metricsSystem,
             syncState,
             Wei.of(1),
-            TransactionPoolConfiguration.builder().build(),
+            TransactionPoolConfiguration.DEFAULT,
             true,
             Optional.empty());
     ethProtocolManager =
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/task/BufferedGetPooledTransactionsFromPeerFetcherTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/task/BufferedGetPooledTransactionsFromPeerFetcherTest.java
new file mode 100644
index 00000000000..fab46ebd98f
--- /dev/null
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/task/BufferedGetPooledTransactionsFromPeerFetcherTest.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * 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.eth.manager.task;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
+import org.hyperledger.besu.ethereum.core.Hash;
+import org.hyperledger.besu.ethereum.core.Transaction;
+import org.hyperledger.besu.ethereum.eth.manager.EthContext;
+import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionsMessageProcessor;
+import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
+import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
+import org.hyperledger.besu.plugin.services.MetricsSystem;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class BufferedGetPooledTransactionsFromPeerFetcherTest {
+
+  @Mock EthPeer ethPeer;
+  @Mock PendingTransactionsMessageProcessor processor;
+  @Mock TransactionPool transactionPool;
+  @Mock EthContext ethContext;
+  @Mock EthScheduler ethScheduler;
+
+  @InjectMocks BufferedGetPooledTransactionsFromPeerFetcher fetcher;
+
+  private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
+
+  private final BlockDataGenerator generator = new BlockDataGenerator();
+
+  @Before
+  public void setup() {
+    when(processor.getTransactionPool()).thenReturn(transactionPool);
+    when(processor.getMetricsSystem()).thenReturn(metricsSystem);
+    when(processor.getEthContext()).thenReturn(ethContext);
+    when(ethContext.getScheduler()).thenReturn(ethScheduler);
+  }
+
+  @Test
+  public void requestTransactionShouldStartTaskWhenUnknownTransaction() {
+
+    final Hash hash = generator.transaction().getHash();
+    final List<Transaction> taskResult = Collections.singletonList(Transaction.builder().build());
+    final AbstractPeerTask.PeerTaskResult<List<Transaction>> peerTaskResult =
+        new AbstractPeerTask.PeerTaskResult<>(ethPeer, taskResult);
+    when(ethScheduler.scheduleSyncWorkerTask(any(GetPooledTransactionsFromPeerTask.class)))
+        .thenReturn(CompletableFuture.completedFuture(peerTaskResult));
+
+    fetcher.addHash(hash);
+    fetcher.requestTransactions();
+
+    verify(ethScheduler).scheduleSyncWorkerTask(any(GetPooledTransactionsFromPeerTask.class));
+    verifyNoMoreInteractions(ethScheduler);
+
+    verify(transactionPool, times(1)).addRemoteTransactions(taskResult);
+  }
+
+  @Test
+  public void requestTransactionShouldSplitRequestIntoSeveralTasks() {
+    for (int i = 0; i < 257; i++) {
+      fetcher.addHash(generator.transaction().getHash());
+    }
+    final AbstractPeerTask.PeerTaskResult<List<Transaction>> peerTaskResult =
+        new AbstractPeerTask.PeerTaskResult<>(ethPeer, new ArrayList<>());
+    when(ethScheduler.scheduleSyncWorkerTask(any(GetPooledTransactionsFromPeerTask.class)))
+        .thenReturn(CompletableFuture.completedFuture(peerTaskResult));
+
+    fetcher.requestTransactions();
+
+    verify(ethScheduler, times(2))
+        .scheduleSyncWorkerTask(any(GetPooledTransactionsFromPeerTask.class));
+    verifyNoMoreInteractions(ethScheduler);
+  }
+
+  @Test
+  public void requestTransactionShouldNotStartTaskWhenTransactionAlreadyInPool() {
+
+    final Hash hash = generator.transaction().getHash();
+    when(transactionPool.getTransactionByHash(hash))
+        .thenReturn(Optional.of(Transaction.builder().build()));
+
+    fetcher.addHash(hash);
+    fetcher.requestTransactions();
+
+    verifyNoInteractions(ethScheduler);
+    verify(transactionPool, never()).addRemoteTransactions(anyList());
+  }
+}
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessorTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessorTest.java
index 2c15234bdb2..d227cea4610 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessorTest.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/PendingTransactionsMessageProcessorTest.java
@@ -18,7 +18,10 @@
 import static java.time.Duration.ofMinutes;
 import static java.time.Instant.now;
 import static java.util.Arrays.asList;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -26,16 +29,24 @@
 
 import org.hyperledger.besu.ethereum.core.BlockDataGenerator;
 import org.hyperledger.besu.ethereum.core.Hash;
+import org.hyperledger.besu.ethereum.core.Transaction;
+import org.hyperledger.besu.ethereum.eth.manager.EthContext;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.messages.NewPooledTransactionHashesMessage;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
+import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactionsMessageProcessor.FetcherCreatorTask;
+import org.hyperledger.besu.plugin.services.MetricsSystem;
 import org.hyperledger.besu.plugin.services.metrics.Counter;
 
+import java.time.Duration;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.Optional;
 
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 
@@ -43,17 +54,38 @@
 public class PendingTransactionsMessageProcessorTest {
 
   @Mock private TransactionPool transactionPool;
+  @Mock private TransactionPoolConfiguration transactionPoolConfiguration;
   @Mock private PeerPendingTransactionTracker transactionTracker;
   @Mock private Counter totalSkippedTransactionsMessageCounter;
   @Mock private EthPeer peer1;
+  @Mock private MetricsSystem metricsSystem;
   @Mock private SyncState syncState;
-  @InjectMocks private PendingTransactionsMessageProcessor messageHandler;
+  @Mock private EthContext ethContext;
+  @Mock private EthScheduler ethScheduler;
+
+  private PendingTransactionsMessageProcessor messageHandler;
 
   private final BlockDataGenerator generator = new BlockDataGenerator();
   private final Hash hash1 = generator.transaction().getHash();
   private final Hash hash2 = generator.transaction().getHash();
   private final Hash hash3 = generator.transaction().getHash();
 
+  @Before
+  public void setup() {
+    when(transactionPoolConfiguration.getEth65TrxAnnouncedBufferingPeriod())
+        .thenReturn(Duration.ofMillis(500));
+    messageHandler =
+        new PendingTransactionsMessageProcessor(
+            transactionTracker,
+            transactionPool,
+            transactionPoolConfiguration,
+            totalSkippedTransactionsMessageCounter,
+            ethContext,
+            metricsSystem,
+            syncState);
+    when(ethContext.getScheduler()).thenReturn(ethScheduler);
+  }
+
   @Test
   public void shouldMarkAllReceivedTransactionsAsSeen() {
     messageHandler.processNewPooledTransactionHashesMessage(
@@ -76,9 +108,35 @@ public void shouldAddInitiatedRequestingTransactions() {
         NewPooledTransactionHashesMessage.create(asList(hash1, hash2, hash3)),
         now(),
         ofMinutes(1));
+
     verify(transactionPool).addTransactionHash(hash1);
     verify(transactionPool).addTransactionHash(hash2);
     verify(transactionPool).addTransactionHash(hash3);
+    verify(transactionPool).getTransactionByHash(hash1);
+    verify(transactionPool).getTransactionByHash(hash2);
+    verify(transactionPool).getTransactionByHash(hash3);
+    verifyNoMoreInteractions(transactionPool);
+  }
+
+  @Test
+  public void shouldNotAddAlreadyPresentTransactions() {
+    when(syncState.isInSync(anyLong())).thenReturn(true);
+
+    when(transactionPool.getTransactionByHash(hash1))
+        .thenReturn(Optional.of(Transaction.builder().build()));
+    when(transactionPool.getTransactionByHash(hash2))
+        .thenReturn(Optional.of(Transaction.builder().build()));
+
+    messageHandler.processNewPooledTransactionHashesMessage(
+        peer1,
+        NewPooledTransactionHashesMessage.create(asList(hash1, hash2, hash3)),
+        now(),
+        ofMinutes(1));
+
+    verify(transactionPool).addTransactionHash(hash3);
+    verify(transactionPool).getTransactionByHash(hash1);
+    verify(transactionPool).getTransactionByHash(hash2);
+    verify(transactionPool).getTransactionByHash(hash3);
     verifyNoMoreInteractions(transactionPool);
   }
 
@@ -117,4 +175,42 @@ public void shouldNotAddReceivedTransactionsToTransactionPoolIfExpired() {
     verify(totalSkippedTransactionsMessageCounter).inc(1);
     verifyNoMoreInteractions(totalSkippedTransactionsMessageCounter);
   }
+
+  @Test
+  public void shouldScheduleGetPooledTransactionsTaskWhenNewTransactionAdded() {
+    when(syncState.isInSync(anyLong())).thenReturn(true);
+
+    final EthScheduler ethScheduler = mock(EthScheduler.class);
+    when(ethContext.getScheduler()).thenReturn(ethScheduler);
+    when(transactionPool.addTransactionHash(hash1)).thenReturn(true);
+
+    messageHandler.processNewPooledTransactionHashesMessage(
+        peer1, NewPooledTransactionHashesMessage.create(asList(hash1, hash2)), now(), ofMinutes(1));
+
+    verify(ethScheduler, times(1))
+        .scheduleFutureTask(any(FetcherCreatorTask.class), any(Duration.class));
+  }
+
+  @Test
+  public void shouldNotScheduleGetPooledTransactionsTaskTwice() {
+    when(syncState.isInSync(anyLong())).thenReturn(true);
+
+    when(transactionPool.addTransactionHash(hash1)).thenReturn(true);
+    when(transactionPool.addTransactionHash(hash2)).thenReturn(true);
+
+    messageHandler.processNewPooledTransactionHashesMessage(
+        peer1,
+        NewPooledTransactionHashesMessage.create(Collections.singletonList(hash1)),
+        now(),
+        ofMinutes(1));
+
+    messageHandler.processNewPooledTransactionHashesMessage(
+        peer1,
+        NewPooledTransactionHashesMessage.create(Collections.singletonList(hash2)),
+        now(),
+        ofMinutes(1));
+
+    verify(ethScheduler, times(1))
+        .scheduleFutureTask(any(FetcherCreatorTask.class), any(Duration.class));
+  }
 }
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TestNode.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TestNode.java
index 8c6d726bb86..ca00e7f8955 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TestNode.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TestNode.java
@@ -136,7 +136,7 @@ public TestNode(
             metricsSystem,
             syncState,
             Wei.ZERO,
-            TransactionPoolConfiguration.builder().build(),
+            TransactionPoolConfiguration.DEFAULT,
             true,
             Optional.empty());
 
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactoryTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactoryTest.java
index a26bf8e06e1..cb7948c076c 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactoryTest.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolFactoryTest.java
@@ -90,12 +90,13 @@ public void testDisconnect() {
             new NoOpMetricsSystem(),
             state,
             Wei.of(1),
-            new TransactionPoolConfiguration(
+            ImmutableTransactionPoolConfiguration.of(
                 1,
                 1,
                 1,
                 1,
                 TransactionPoolConfiguration.DEFAULT_PRICE_BUMP,
+                TransactionPoolConfiguration.ETH65_TRX_ANNOUNCED_BUFFERING_PERIOD,
                 TransactionPoolConfiguration.DEFAULT_RPC_TX_FEE_CAP),
             pendingTransactions,
             peerTransactionTracker,
@@ -175,12 +176,13 @@ public void testNoEth65() {
             new NoOpMetricsSystem(),
             state,
             Wei.of(1),
-            new TransactionPoolConfiguration(
+            ImmutableTransactionPoolConfiguration.of(
                 1,
                 1,
                 1,
                 1,
                 TransactionPoolConfiguration.DEFAULT_PRICE_BUMP,
+                TransactionPoolConfiguration.ETH65_TRX_ANNOUNCED_BUFFERING_PERIOD,
                 TransactionPoolConfiguration.DEFAULT_RPC_TX_FEE_CAP),
             pendingTransactions,
             peerTransactionTracker,
diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolTest.java
index d57be38928f..9cda887e0fe 100644
--- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolTest.java
+++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/transactions/TransactionPoolTest.java
@@ -683,7 +683,7 @@ public void shouldIgnoreFeeCapIfSetZero() {
             Wei.ZERO,
             metricsSystem,
             Optional.empty(),
-            TransactionPoolConfiguration.builder().txFeeCap(Wei.ZERO).build());
+            ImmutableTransactionPoolConfiguration.builder().txFeeCap(Wei.ZERO).build());
     when(transactionValidator.validate(any(Transaction.class), any(Optional.class)))
         .thenReturn(valid());
     when(transactionValidator.validateForSender(
@@ -722,7 +722,7 @@ public void shouldRejectLocalTransactionIfFeeCapExceeded() {
             Wei.ZERO,
             metricsSystem,
             Optional.empty(),
-            TransactionPoolConfiguration.builder().txFeeCap(twoEthers).build());
+            ImmutableTransactionPoolConfiguration.builder().txFeeCap(twoEthers).build());
 
     final TransactionTestFixture builder = new TransactionTestFixture();
     final Transaction transactionLocal =
diff --git a/ethereum/retesteth/src/main/java/org/hyperledger/besu/ethereum/retesteth/RetestethContext.java b/ethereum/retesteth/src/main/java/org/hyperledger/besu/ethereum/retesteth/RetestethContext.java
index 6797c65bce9..de83908ccf7 100644
--- a/ethereum/retesteth/src/main/java/org/hyperledger/besu/ethereum/retesteth/RetestethContext.java
+++ b/ethereum/retesteth/src/main/java/org/hyperledger/besu/ethereum/retesteth/RetestethContext.java
@@ -37,6 +37,7 @@
 import org.hyperledger.besu.ethereum.eth.manager.EthPeers;
 import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
 import org.hyperledger.besu.ethereum.eth.sync.state.SyncState;
+import org.hyperledger.besu.ethereum.eth.transactions.ImmutableTransactionPoolConfiguration;
 import org.hyperledger.besu.ethereum.eth.transactions.PendingTransactions;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;
 import org.hyperledger.besu.ethereum.eth.transactions.TransactionPoolConfiguration;
@@ -190,7 +191,7 @@ private boolean buildContext(
     final EthContext ethContext = new EthContext(ethPeers, new EthMessages(), ethScheduler);
 
     final TransactionPoolConfiguration transactionPoolConfiguration =
-        TransactionPoolConfiguration.builder().build();
+        ImmutableTransactionPoolConfiguration.builder().build();
 
     transactionPool =
         TransactionPoolFactory.createTransactionPool(