then:
        def e = thrown(GradleException)
        e.message == "Cannot add incoming message handler after connection established."

        cleanup:
        connection?.stop()
        connectionBuilder?.stop()
    }

    def "cannot create outgoing message transmitter on established connections"() {
        def connection = new TestConnection()

        given:
        connectCompletion.create(_) >> connection

		apiextensionshelpers.IsCRDConditionTrue(cachedCRD, apiextensionsv1.Established) {
		return nil
	}

	crd := cachedCRD.DeepCopy()
	establishedCondition := apiextensionsv1.CustomResourceDefinitionCondition{
		Type:    apiextensionsv1.Established,
		Status:  apiextensionsv1.ConditionTrue,
		Reason:  "InitialNamesAccepted",
		Message: "the initial names have been accepted",

  open fun connectStart(
    route: Route,
    call: Call,
  ) {}

  /**
   * Invoked when a connection fails to be established.
   */
  open fun connectFailed(
    route: Route,
    call: Call,
    failure: IOException,
  ) {}

  /**
   * Invoked as soon as a connection is successfully established.
   */
  open fun connectEnd(
    connection: Connection,
    route: Route,
    call: Call,
  ) {}

  /**

				-A KUBE-FORWARD -m comment --comment "kubernetes forwarding rules" -m mark --mark 0x4000/0x4000 -j ACCEPT
				-A KUBE-FORWARD -m comment --comment "kubernetes forwarding conntrack rule" -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT
				COMMIT
				*nat
				:KUBE-SERVICES - [0:0]
				:KUBE-NODEPORTS - [0:0]
				:KUBE-POSTROUTING - [0:0]
				:KUBE-MARK-MASQ - [0:0]
				:KUBE-SVC-XPGD46QRK7WJZT7O - [0:0]

apiVersion: release-notes/v2
kind: feature
area: installation
issue:
  - 47257
releaseNotes:
  - |
    **Added** eBPF redirection support for ambient after CNCF established guidance around dual-licensed eBPF bytecode.

iptables -t mangle -A PREROUTING -p tcp -j ISTIO_INBOUND
iptables -t mangle -A ISTIO_INBOUND -p tcp --dport 32000 -m conntrack --ctstate RELATED,ESTABLISHED -j ISTIO_DIVERT
iptables -t mangle -A ISTIO_INBOUND -p tcp --dport 32000 -j ISTIO_TPROXY
iptables -t mangle -A ISTIO_INBOUND -p tcp --dport 31000 -m conntrack --ctstate RELATED,ESTABLISHED -j ISTIO_DIVERT
iptables -t mangle -A ISTIO_INBOUND -p tcp --dport 31000 -j ISTIO_TPROXY

iptables -t mangle -A ISTIO_PRERT -p tcp -m tcp --dport 15008 -m mark ! --mark 0x539/0xfff -j TPROXY --on-port 15008 --tproxy-mark 0x111/0xfff
iptables -t mangle -A ISTIO_PRERT -p tcp -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT
iptables -t mangle -A ISTIO_PRERT ! -d 127.0.0.1/32 -p tcp -m mark ! --mark 0x539/0xfff -j TPROXY --on-port 15006 --tproxy-mark 0x111/0xfff

 */
@ServiceScope(Scope.Global.class)
public interface MessagingClient {
    /**
     * Creates a connection to the given address. Blocks until the connection has been established.
     *
     * @param address The address to connect to.
     */
    ObjectConnection getConnection(Address address);

	// which we had if we set Established to true here.
	establishedCondition := apiextensionsv1.CustomResourceDefinitionCondition{
		Type:    apiextensionsv1.Established,
		Status:  apiextensionsv1.ConditionFalse,
		Reason:  "NotAccepted",
		Message: "not all names are accepted",
	}
	if old := apiextensionshelpers.FindCRDCondition(in, apiextensionsv1.Established); old != nil {
		establishedCondition = *old
	}

				case watch.Added, watch.Modified, watch.Deleted:
					// all expected
				default:
					t.Errorf("unexpected watch event: %#v", event)
				}
			}
		}(i)
	}

	// Let all the established watches soak request loops soak
	time.Sleep(5 * time.Second)

	// Update CRD and ensure that all watches are gracefully terminated.
	updateCRD()

	drained := make(chan struct{})
	go func() {