0 * _

        when:
        broadcast.something("12")

        then:
        1 * listener.something("12")
        0 * _
    }

    def "creates stateful listener registered after broadcaster is created"() {
        def created = Mock(Runnable)
        def listener = Mock(TestListener)

        when:
        def broadcast = listenerManager.getBroadcaster(TestListener)

	broadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: client.EventsV1()})
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	informerFactory := informers.NewSharedInformerFactory(client, 0)
	sched, err := New(
		ctx,
		client,
		informerFactory,
		nil,
		profile.NewRecorderFactory(broadcaster),
		WithProfiles(

  "tf_device.cluster"() ({
    // CHECK-NOT: ResourceScatterUpdate
    // CHECK-NOT: BroadcastTo
    // CHECK: TensorScatterAdd
    "tf.ResourceScatterAdd"(%resource, %indices, %updates) {device = ""} : (tensor<*x!tf_type.resource>, tensor<?xi32>, tensor<i32>) -> ()
    tf_device.return
  }) : () -> ()

                       std::vector<Output>* grad_outputs) {
  if (grad_inputs.size() != 1) {
    return errors::InvalidArgument("BroadcastTo grad should have 1 grad input");
  }
  if (op.num_inputs() != 2) {
    return errors::InvalidArgument("BroadcastTo requires 2 inputs");
  }

  auto x_shape = Shape(scope, op.input(0));
  auto args = internal::BroadcastGradientArgs(scope, x_shape, op.input(1));

// CHECK: %[[BROADCAST:.+]] = "tfl.broadcast_to"(%[[ARG0]], %[[SHAPE]]) : (tensor<1x2x!quant.uniform<i8:f32, 2.000000e+00:3>>, tensor<2xi32>) -> tensor<3x2x!quant.uniform<i8:f32, 2.000000e+00:3>>
// CHECK: return %[[BROADCAST]]

// -----

// Tests that a quantized `stablehlo.broadcast_in_dim` is converted to
// `tfl.transpose` and `tfl.broadcast_to` when `broadcast_dimensions` is not in
// ascending order.

	endpointsInformer coreinformers.EndpointsInformer, client clientset.Interface, endpointUpdatesBatchPeriod time.Duration) *Controller {
	broadcaster := record.NewBroadcaster(record.WithContext(ctx))
	recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "endpoint-controller"})

	e := &Controller{
		client: client,
		queue: workqueue.NewTypedRateLimitingQueueWithConfig(

	defer utilruntime.HandleCrash()

	// Start events processing pipeline.
	nc.broadcaster.StartStructuredLogging(3)
	logger := klog.FromContext(ctx)
	logger.Info("Sending events to api server")
	nc.broadcaster.StartRecordingToSink(
		&v1core.EventSinkImpl{
			Interface: v1core.New(nc.kubeClient.CoreV1().RESTClient()).Events(""),
		})
	defer nc.broadcaster.Shutdown()

	// Close node update queue to cleanup go routine.

    // Reshaping the gradient (wrt "ac") to [2,1,3,1] and broadcasting it to
    // the shape [2,5,3,4] results in the gradient wrt "abcd".
    return BroadcastTo(scope, Reshape(scope, output_grad, reduced_shape),
                       input_shape);
  }

  // If we *do* have traces or transpose operations, then prepend the extra

			nodePortAddresses = append(nodePortAddresses, fmt.Sprintf("%s/128", nodeIP.String()))
		}
		config.NodePortAddresses = nodePortAddresses
	}

	s.Broadcaster = events.NewBroadcaster(&events.EventSinkImpl{Interface: s.Client.EventsV1()})
	s.Recorder = s.Broadcaster.NewRecorder(proxyconfigscheme.Scheme, "kube-proxy")

	s.NodeRef = &v1.ObjectReference{
		Kind:      "Node",
		Name:      s.Hostname,

    def listenerManager = new DefaultListenerManager(Scope.BuildSession)
    def buildOperationListenerManager = new DefaultBuildOperationListenerManager()
    def broadcast = buildOperationListenerManager.broadcaster
    def listener = Mock(BuildOperationNotificationListener)

    BuildOperationNotificationBridge bridgeInstance

    def "removes listener when stopped"() {
        given:
        def buildOperationListener