Adapt EC for eventing manager

components/eventing-controller/cmd/eventing-controller/main.go

@@ -118,7 +118,8 @@ func main() {
 	// Start the backend manager.
 	ctx := context.Background()
 	recorder := mgr.GetEventRecorderFor("backend-controller")
-	backendReconciler := backend.NewReconciler(ctx, natsSubMgr, natsConfig, envConfig, bebSubMgr,
+	backendConfig := env.GetBackendConfig()
+	backendReconciler := backend.NewReconciler(ctx, natsSubMgr, natsConfig, envConfig, backendConfig, bebSubMgr,
 		mgr.GetClient(), ctrLogger, recorder)
 	if err = backendReconciler.SetupWithManager(mgr); err != nil {
 		setupLogger.Fatalw("Failed to start backend controller", "error", err)

components/eventing-controller/controllers/backend/reconciler.go

@@ -103,10 +103,16 @@ type Reconciler struct {
 	credentials oauth2Credentials
 }
 
-func NewReconciler(ctx context.Context, natsSubMgr subscriptionmanager.Manager, natsConfig env.NATSConfig,
-	envCfg env.Config, bebSubMgr subscriptionmanager.Manager, client client.Client, logger *logger.Logger,
+func NewReconciler(
+	ctx context.Context,
+	natsSubMgr subscriptionmanager.Manager,
+	natsConfig env.NATSConfig,
+	envCfg env.Config,
+	backendCfg env.BackendConfig,
+	bebSubMgr subscriptionmanager.Manager,
+	client client.Client,
+	logger *logger.Logger,
 	recorder record.EventRecorder) *Reconciler {
-	cfg := env.GetBackendConfig()
 	return &Reconciler{
 		ctx:        ctx,
 		natsSubMgr: natsSubMgr,
@@ -116,10 +122,18 @@ func NewReconciler(ctx context.Context, natsSubMgr subscriptionmanager.Manager,
 		Client:     client,
 		logger:     logger,
 		record:     recorder,
-		cfg:        cfg,
+		cfg:        backendCfg,
 	}
 }
 
+func (r *Reconciler) SetNatsConfig(natsConfig env.NATSConfig) {
+	r.natsConfig = natsConfig
+}
+
+func (r *Reconciler) SetBackendConfig(backendCfg env.BackendConfig) {
+	r.cfg = backendCfg
+}
+
 // +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;update;patch;create;delete
 // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;update;patch;create;delete
 // +kubebuilder:rbac:groups=eventing.kyma-project.io,resources=eventingbackends,verbs=get;list;watch;create;update;patch;delete
@@ -622,7 +636,16 @@ func getSecretStringData(clientID, clientSecret, tokenEndpoint, grantType, publi
 }
 
 func (r *Reconciler) CreateOrUpdatePublisherProxy(ctx context.Context, backend eventingv1alpha1.BackendType) (*appsv1.Deployment, error) {
+	return r.CreateOrUpdatePublisherProxyDeployment(ctx, backend, true)
+}
+
+func (r *Reconciler) CreateOrUpdatePublisherProxyDeployment(
+	ctx context.Context,
+	backend eventingv1alpha1.BackendType,
+	withOwnerReference bool) (*appsv1.Deployment, error) {
 	var desiredPublisher *appsv1.Deployment
+	// set backend type here so that the function can be used in eventing-manager
+	r.backendType = backend
 
 	switch backend {
 	case eventingv1alpha1.NatsBackendType:
@@ -633,8 +656,10 @@ func (r *Reconciler) CreateOrUpdatePublisherProxy(ctx context.Context, backend e
 		return nil, fmt.Errorf("unknown EventingBackend type %q", backend)
 	}
 
-	if err := r.setAsOwnerReference(ctx, desiredPublisher); err != nil {
-		return nil, errors.Wrapf(err, "set owner reference for Event Publisher failed")
+	if withOwnerReference {
+		if err := r.setAsOwnerReference(ctx, desiredPublisher); err != nil {
+			return nil, errors.Wrapf(err, "set owner reference for Event Publisher failed")
+		}
 	}
 
 	currentPublisher, err := r.getEPPDeployment(ctx)

components/eventing-controller/controllers/backend/reconciler_internal_integration_test.go

@@ -169,11 +169,13 @@ var _ = BeforeSuite(func(done Done) {
 	}
 
 	envConfig := env.Config{}
+	backendConfig := env.GetBackendConfig()
 	err = NewReconciler(
 		context.Background(),
 		natsSubMgr,
 		natsConfig,
 		envConfig,
+		backendConfig,
 		bebSubMgr,
 		k8sManager.GetClient(),
 		defaultLogger,

components/eventing-controller/pkg/backend/jetstream/utils.go

@@ -148,13 +148,14 @@ func getStreamConfig(natsConfig env.NATSConfig) (*nats.StreamConfig, error) {
 	}
 
 	streamConfig := &nats.StreamConfig{
-		Name:      natsConfig.JSStreamName,
-		Storage:   storage,
-		Replicas:  natsConfig.JSStreamReplicas,
-		Retention: retentionPolicy,
-		MaxMsgs:   natsConfig.JSStreamMaxMessages,
-		MaxBytes:  maxBytes.Value(),
-		Discard:   discardPolicy,
+		Name:              natsConfig.JSStreamName,
+		Storage:           storage,
+		Replicas:          natsConfig.JSStreamReplicas,
+		Retention:         retentionPolicy,
+		MaxMsgs:           natsConfig.JSStreamMaxMessages,
+		MaxBytes:          maxBytes.Value(),
+		Discard:           discardPolicy,
+		MaxMsgsPerSubject: natsConfig.JSStreamMaxMsgsPerTopic,
 		// Since one stream is used to store events of all types, the stream has to match all event types, and therefore
 		// we use the wildcard char >. However, to avoid matching internal JetStream and non-Kyma-related subjects, we
 		// use a prefix. This prefix is handled only on the JetStream level (i.e. JetStream handler

components/eventing-controller/pkg/env/nats_config.go

@@ -39,6 +39,7 @@ type NATSConfig struct {
 	JSStreamRetentionPolicy string `envconfig:"JS_STREAM_RETENTION_POLICY" default:"interest"`
 	JSStreamMaxMessages     int64  `envconfig:"JS_STREAM_MAX_MSGS" default:"-1"`
 	JSStreamMaxBytes        string `envconfig:"JS_STREAM_MAX_BYTES" default:"-1"`
+	JSStreamMaxMsgsPerTopic int64  `envconfig:"JS_STREAM_MAX_MSGS_PER_TOPIC" default:"-1"`
 	// JSStreamDiscardPolicy specifies which events to discard from the stream in case limits are reached
 	//  new: reject new messages for the stream
 	//  old: discard old messages from the stream to make room for new messages