Fix crash related to dynamic updates to UDP proxy via CDS/EDS

source/extensions/filters/udp/udp_proxy/udp_proxy_filter.cc

@@ -49,11 +49,11 @@ void UdpProxyFilter::onClusterAddOrUpdate(absl::string_view cluster_name,
          &cluster_infos_[cluster_name]->cluster_ != &cluster);
 
   if (config_->usingPerPacketLoadBalancing()) {
-    cluster_infos_.emplace(cluster_name,
-                           std::make_unique<PerPacketLoadBalancingClusterInfo>(*this, cluster));
+    cluster_infos_.insert_or_assign(
+        cluster_name, std::make_unique<PerPacketLoadBalancingClusterInfo>(*this, cluster));
   } else {
-    cluster_infos_.emplace(cluster_name,
-                           std::make_unique<StickySessionClusterInfo>(*this, cluster));
+    cluster_infos_.insert_or_assign(cluster_name,
+                                    std::make_unique<StickySessionClusterInfo>(*this, cluster));
   }
 }
 

test/extensions/filters/udp/udp_proxy/udp_proxy_filter_test.cc

@@ -826,6 +826,51 @@ stat_prefix: foo
   EXPECT_EQ(0, config_->stats().downstream_sess_active_.value());
 }
 
+// Test updates to existing cluster (e.g. priority set changes, etc)
+TEST_F(UdpProxyFilterTest, ClusterDynamicInfoMapUpdate) {
+  InSequence s;
+
+  setup(readConfig(R"EOF(
+stat_prefix: foo
+matcher:
+  on_no_match:
+    action:
+      name: route
+      typed_config:
+        '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.Route
+        cluster: fake_cluster
+  )EOF"),
+        false);
+
+  // Initial ThreadLocalCluster, scoped lifetime
+  // mimics replacement of old ThreadLocalCluster via postThreadLocalClusterUpdate
+  {
+    NiceMock<Upstream::MockThreadLocalCluster> other_thread_local_cluster;
+    other_thread_local_cluster.cluster_.info_->name_ = "fake_cluster";
+    Upstream::ThreadLocalClusterCommand command =
+        [&other_thread_local_cluster]() -> Upstream::ThreadLocalCluster& {
+      return other_thread_local_cluster;
+    };
+    cluster_update_callbacks_->onClusterAddOrUpdate(other_thread_local_cluster.info()->name(),
+                                                    command);
+  }
+
+  // Push new cluster (getter), we expect this to result in new ClusterInfos object in map
+  {
+    Upstream::ThreadLocalClusterCommand command = [this]() -> Upstream::ThreadLocalCluster& {
+      return factory_context_.server_factory_context_.cluster_manager_.thread_local_cluster_;
+    };
+    cluster_update_callbacks_->onClusterAddOrUpdate(
+        factory_context_.server_factory_context_.cluster_manager_.thread_local_cluster_.info()
+            ->name(),
+        command);
+  }
+
+  expectSessionCreate(upstream_address_);
+  test_sessions_[0].expectWriteToUpstream("hello", 0, nullptr, true);
+  recvDataFromDownstream("10.0.0.1:1000", "10.0.0.2:80", "hello");
+}
+
 // Hitting the maximum per-cluster connection/session circuit breaker.
 TEST_F(UdpProxyFilterTest, MaxSessionsCircuitBreaker) {
   InSequence s;