diff --git a/.changeset/deep-aliens-camp.md b/.changeset/deep-aliens-camp.md
new file mode 100644
index 0000000000..ca5c3d8ed4
--- /dev/null
+++ b/.changeset/deep-aliens-camp.md
@@ -0,0 +1,7 @@
+---
+'@backstage/plugin-events-backend-module-kafka': minor
+---
+
+Adds a new module `kafka` for plugin-events-backend
+
+The module introduces the `KafkaConsumerClient` which creates a Kafka client used to establish consumer connections. It also provides the `KafkaConsumingEventPublisher`, a consumer that subscribes to configured Kafka topics and publishes received messages to the Event Service.
diff --git a/plugins/events-backend-module-kafka/.eslintrc.js b/plugins/events-backend-module-kafka/.eslintrc.js
new file mode 100644
index 0000000000..e2a53a6ad2
--- /dev/null
+++ b/plugins/events-backend-module-kafka/.eslintrc.js
@@ -0,0 +1 @@
+module.exports = require('@backstage/cli/config/eslint-factory')(__dirname);
diff --git a/plugins/events-backend-module-kafka/README.md b/plugins/events-backend-module-kafka/README.md
new file mode 100644
index 0000000000..dbc5bf5607
--- /dev/null
+++ b/plugins/events-backend-module-kafka/README.md
@@ -0,0 +1,78 @@
+# @backstage/backstage-plugin-events-backend-module-kafka
+
+Welcome to the `events-backend-module-kafka` backend module!
+
+This package is a module for the `events-backend` backend plugin and extends the events system with an `KafkaConsumingEventPublisher.`
+
+This event publisher will allow you to receive events from an Kafka queue and will publish these to the used `EventsService` implementation.
+
+## Configuration
+
+To set up Kafka queues, you need to configure the following values:
+
+```yaml
+events:
+  modules:
+    kafka:
+      kafkaConsumingEventPublisher:
+        clientId: your-client-id # (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
+        brokers: # (Required) List of brokers in the Kafka cluster to connect to.
+          - broker1
+          - broker2
+        topics:
+          - topic: 'backstage.topic' # (Required) Replace with actual topic name as expected by subscribers
+            kafka:
+              topics: # (Required) The Kafka topics to subscribe to.
+                - topic1
+              groupId: your-group-id # (Required) The GroupId to be used by the topic consumers.
+```
+
+For a complete list of all available fields that can be configured, refer to the [config.d.ts file](./config.d.ts).
+
+### Optional SSL Configuration
+
+If your Kafka cluster requires SSL, you can configure it as follows:
+
+```yaml
+events:
+  modules:
+    kafka:
+      kafkaConsumingEventPublisher:
+        ssl:
+          rejectUnauthorized: true # (Optional) If true, the server certificate is verified against the list of supplied CAs.
+          ca: [path/to/ca-cert] # (Optional) Array of trusted certificates in PEM format.
+          key: path/to/client-key # (Optional) Private key in PEM format.
+          cert: path/to/client-cert # (Optional) Public x509 certificate in PEM format.
+```
+
+### Optional SASL Authentication Configuration
+
+If your Kafka cluster requires `SASL` authentication, you can configure it as follows:
+
+```yaml
+events:
+  modules:
+    kafka:
+      kafkaConsumingEventPublisher:
+        sasl:
+          mechanism: 'plain' # SASL mechanism ('plain', 'scram-sha-256' or 'scram-sha-512')
+          username: your-username # SASL username
+          password: your-password # SASL password
+```
+
+This section includes optional `SSL` and `SASL` authentication configuration for enhanced security.
+
+## Installation
+
+1. Install this module
+2. Add your configuration.
+
+```bash
+# From your Backstage root directory
+yarn --cwd packages/backend add @backstage/plugin-events-backend-module-kafka
+```
+
+```typescript
+// packages/backend/src/index.ts
+backend.add(import('@backstage/plugin-events-backend-module-kafka'));
+```
diff --git a/plugins/events-backend-module-kafka/catalog-info.yaml b/plugins/events-backend-module-kafka/catalog-info.yaml
new file mode 100644
index 0000000000..72358ec3c3
--- /dev/null
+++ b/plugins/events-backend-module-kafka/catalog-info.yaml
@@ -0,0 +1,10 @@
+apiVersion: backstage.io/v1alpha1
+kind: Component
+metadata:
+  name: backstage-plugin-events-backend-module-kafka
+  title: '@backstage/plugin-events-backend-module-kafka'
+  description: The kafka backend module for the events plugin.
+spec:
+  lifecycle: experimental
+  type: backstage-backend-plugin-module
+  owner: maintainers
diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts
new file mode 100644
index 0000000000..49c417098c
--- /dev/null
+++ b/plugins/events-backend-module-kafka/config.d.ts
@@ -0,0 +1,205 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import { HumanDuration } from '@backstage/types';
+
+export interface Config {
+  events?: {
+    modules?: {
+      /**
+       * events-backend-module-kafka plugin configuration.
+       */
+      kafka?: {
+        /**
+         * Configuration for KafkaConsumingEventPublisher
+         */
+        kafkaConsumingEventPublisher?: {
+          /**
+           * (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
+           */
+          clientId: string;
+          /**
+           * (Required) List of brokers in the Kafka cluster to connect to.
+           */
+          brokers: string[];
+          /**
+           * Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect.
+           * See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options
+           */
+          ssl?:
+            | {
+                ca?: string[];
+                /** @visibility secret */
+                key?: string;
+                cert?: string;
+                rejectUnauthorized?: boolean;
+              }
+            | boolean;
+          /**
+           * Optional SASL connection parameters.
+           */
+          sasl?: {
+            mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
+            username: string;
+            /** @visibility secret */
+            password: string;
+          };
+
+          /**
+           * Optional retry connection parameters.
+           */
+          retry: {
+            /**
+             * (Optional) Maximum wait time for a retry
+             * Default: 30000 ms.
+             */
+            maxRetryTime: HumanDuration | string;
+
+            /**
+             * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor)
+             * Default: 300 ms.
+             */
+            initialRetryTime: HumanDuration | string;
+
+            /**
+             * (Optional) Randomization factor
+             * Default: 0.2.
+             */
+            factor: number;
+
+            /**
+             * (Optional) Exponential factor
+             * Default: 2.
+             */
+            multiplier: number;
+
+            /**
+             * (Optional) Max number of retries per call
+             * Default: 5.
+             */
+            retries: number;
+          };
+
+          /**
+           * (Optional) Timeout for authentication requests.
+           * Default: 10000 ms.
+           */
+          authenticationTimeout: HumanDuration | string;
+
+          /**
+           * (Optional) Time to wait for a successful connection.
+           * Default: 1000 ms.
+           */
+          connectionTimeout: HumanDuration | string;
+
+          /**
+           * (Optional) Time to wait for a successful request.
+           * Default: 30000 ms.
+           */
+          requestTimeout: HumanDuration | string;
+
+          /**
+           * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false.
+           * Default: true
+           */
+          enforceRequestTimeout: boolean;
+
+          /**
+           * Contains a object per topic for which an Kafka queue
+           * should be used as source of events.
+           */
+          topics: Array<{
+            /**
+             * (Required) The Backstage topic to publish to
+             */
+            topic: string;
+            /**
+             * (Required) KafkaConsumer-related configuration.
+             */
+            kafka: {
+              /**
+               * (Required) The Kafka topics to subscribe to
+               */
+              topics: string[];
+              /**
+               * (Required) The GroupId to be used by the topic consumers
+               */
+              groupId: string;
+
+              /**
+               * (Optional) Timeout used to detect failures.
+               * The consumer sends periodic heartbeats to indicate its liveness to the broker.
+               * If no heartbeats are received by the broker before the expiration of this session timeout,
+               * then the broker will remove this consumer from the group and initiate a rebalance
+               * Default: 30000 ms.
+               */
+              sessionTimeout: HumanDuration | string;
+
+              /**
+               * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group
+               * Default: 60000 ms.
+               */
+              rebalanceTimeout: HumanDuration | string;
+
+              /**
+               * (Optional) The expected time between heartbeats to the consumer coordinator.
+               * Heartbeats are used to ensure that the consumer's session stays active.
+               * The value must be set lower than session timeout
+               * Default: 3000 ms.
+               */
+              heartbeatInterval: HumanDuration | string;
+
+              /**
+               * (Optional) The period of time after which we force a refresh of metadata
+               * even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions
+               * Default: 300000 ms (5 minutes).
+               */
+              metadataMaxAge: HumanDuration | string;
+
+              /**
+               * (Optional) The maximum amount of data per-partition the server will return.
+               * This size must be at least as large as the maximum message size the server allows
+               * or else it is possible for the producer to send messages larger than the consumer can fetch.
+               * If that happens, the consumer can get stuck trying to fetch a large message on a certain partition
+               * Default: 1048576 (1MB)
+               */
+              maxBytesPerPartition: number;
+
+              /**
+               * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate.
+               * Default: 1
+               */
+              minBytes: number;
+
+              /**
+               * (Optional) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0
+               * Default: 10485760 (10MB)
+               */
+              maxBytes: number;
+
+              /**
+               * (Optional) The maximum amount of time the server will block before answering the fetch request
+               * if there isn’t sufficient data to immediately satisfy the requirement given by minBytes
+               * Default: 5000
+               */
+              maxWaitTime: HumanDuration | string;
+            };
+          }>;
+        };
+      };
+    };
+  };
+}
diff --git a/plugins/events-backend-module-kafka/package.json b/plugins/events-backend-module-kafka/package.json
new file mode 100644
index 0000000000..488ecc0a0e
--- /dev/null
+++ b/plugins/events-backend-module-kafka/package.json
@@ -0,0 +1,51 @@
+{
+  "name": "@backstage/plugin-events-backend-module-kafka",
+  "version": "0.0.0",
+  "description": "The kafka backend module for the events plugin.",
+  "backstage": {
+    "role": "backend-plugin-module",
+    "pluginId": "events",
+    "pluginPackage": "@backstage/plugin-events-backend"
+  },
+  "publishConfig": {
+    "access": "public",
+    "main": "dist/index.cjs.js",
+    "types": "dist/index.d.ts"
+  },
+  "homepage": "https://backstage.io",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/backstage/backstage",
+    "directory": "plugins/events-backend-module-kafka"
+  },
+  "license": "Apache-2.0",
+  "main": "src/index.ts",
+  "types": "src/index.ts",
+  "files": [
+    "dist",
+    "config.d.ts"
+  ],
+  "scripts": {
+    "build": "backstage-cli package build",
+    "clean": "backstage-cli package clean",
+    "lint": "backstage-cli package lint",
+    "prepack": "backstage-cli package prepack",
+    "postpack": "backstage-cli package postpack",
+    "start": "backstage-cli package start",
+    "test": "backstage-cli package test"
+  },
+  "dependencies": {
+    "@backstage/backend-plugin-api": "workspace:^",
+    "@backstage/config": "workspace:^",
+    "@backstage/plugin-events-node": "workspace:^",
+    "@backstage/types": "workspace:^",
+    "kafkajs": "^2.2.4",
+    "luxon": "^3.0.0"
+  },
+  "devDependencies": {
+    "@backstage/backend-test-utils": "workspace:^",
+    "@backstage/cli": "workspace:^",
+    "@backstage/plugin-events-backend-test-utils": "workspace:^"
+  },
+  "configSchema": "config.d.ts"
+}
diff --git a/plugins/events-backend-module-kafka/report.api.md b/plugins/events-backend-module-kafka/report.api.md
new file mode 100644
index 0000000000..a6b129b82e
--- /dev/null
+++ b/plugins/events-backend-module-kafka/report.api.md
@@ -0,0 +1,11 @@
+## API Report File for "@backstage/plugin-events-backend-module-kafka"
+
+> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
+
+```ts
+import { BackendFeature } from '@backstage/backend-plugin-api';
+
+// @public
+const eventsModuleKafkaConsumingEventPublisher: BackendFeature;
+export default eventsModuleKafkaConsumingEventPublisher;
+```
diff --git a/plugins/events-backend-module-kafka/src/index.ts b/plugins/events-backend-module-kafka/src/index.ts
new file mode 100644
index 0000000000..3ba6207d7d
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/index.ts
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+/**
+ * The module "kafka" for the Backstage backend plugin "events"
+ * adding an Kafka-based publisher,
+ * receiving events from an Kafka topic and passing it to the
+ * internal event broker.
+ *
+ * @packageDocumentation
+ */
+
+export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher';
diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts
new file mode 100644
index 0000000000..5824bf4d41
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts
@@ -0,0 +1,132 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { KafkaConsumerClient } from './KafkaConsumerClient';
+import { ConfigReader } from '@backstage/config';
+import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher';
+import { mockServices } from '@backstage/backend-test-utils';
+
+jest.mock('kafkajs');
+jest.mock('./KafkaConsumingEventPublisher');
+
+describe('KafkaConsumerClient', () => {
+  const mockLogger = mockServices.logger.mock();
+  const mockEvents = mockServices.events.mock();
+
+  const mockConfig = new ConfigReader({
+    events: {
+      modules: {
+        kafka: {
+          kafkaConsumingEventPublisher: {
+            clientId: 'backstage-events',
+            brokers: ['kafka1:9092', 'kafka2:9092'],
+            topics: [
+              {
+                topic: 'fake1',
+                kafka: {
+                  topics: ['topic-A'],
+                  groupId: 'my-group',
+                },
+              },
+              {
+                topic: 'fake2',
+                kafka: {
+                  topics: ['topic-B'],
+                  groupId: 'my-group',
+                },
+              },
+            ],
+          },
+        },
+      },
+    },
+  });
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('should create an instance from config', () => {
+    const client = KafkaConsumerClient.fromConfig({
+      config: mockConfig,
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    expect(client).toBeInstanceOf(KafkaConsumerClient);
+  });
+
+  it('should not create an instance from config', () => {
+    const client = KafkaConsumerClient.fromConfig({
+      config: new ConfigReader({}),
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    expect(client).toBeUndefined();
+  });
+
+  it('should create a consumer for each topic from config', () => {
+    KafkaConsumerClient.fromConfig({
+      config: mockConfig,
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    expect(KafkaConsumingEventPublisher.fromConfig).toHaveBeenCalledTimes(2);
+  });
+
+  it('should start all consumers', async () => {
+    const mockConsumer = {
+      start: jest.fn().mockResolvedValue(undefined),
+    };
+    (KafkaConsumingEventPublisher.fromConfig as jest.Mock).mockReturnValue(
+      mockConsumer,
+    );
+
+    const client = KafkaConsumerClient.fromConfig({
+      config: mockConfig,
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    expect(client).toBeDefined();
+
+    await client?.start();
+
+    expect(mockConsumer.start).toHaveBeenCalled();
+  });
+
+  it('should shutdown all consumers', async () => {
+    const mockConsumer = {
+      shutdown: jest.fn().mockResolvedValue(undefined),
+    };
+    (KafkaConsumingEventPublisher.fromConfig as jest.Mock).mockReturnValue(
+      mockConsumer,
+    );
+
+    const client = KafkaConsumerClient.fromConfig({
+      config: mockConfig,
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    expect(client).toBeDefined();
+
+    await client?.shutdown();
+
+    expect(mockConsumer.shutdown).toHaveBeenCalled();
+  });
+});
diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts
new file mode 100644
index 0000000000..735b0fc224
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { LoggerService } from '@backstage/backend-plugin-api';
+import { Config } from '@backstage/config';
+import { EventsService } from '@backstage/plugin-events-node';
+import { Kafka } from 'kafkajs';
+import { KafkaEventSourceConfig, readConfig } from './config';
+import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher';
+import { loggerServiceAdapter } from './LoggerServiceAdapter';
+
+/**
+ * KafkaConsumerClient
+ *
+ * This class creates the Kafka client that will be used to create the KafkaConsumingEventPublisher
+ *
+ * @public
+ */
+export class KafkaConsumerClient {
+  private readonly kafka: Kafka;
+  private readonly consumers: KafkaConsumingEventPublisher[];
+
+  static fromConfig(options: {
+    config: Config;
+    events: EventsService;
+    logger: LoggerService;
+  }): KafkaConsumerClient | undefined {
+    const kafkaConfig = readConfig(options.config);
+
+    if (!kafkaConfig) {
+      options.logger.info(
+        'Kafka consumer not configured, skipping initialization',
+      );
+      return undefined;
+    }
+
+    return new KafkaConsumerClient(options.logger, options.events, kafkaConfig);
+  }
+
+  private constructor(
+    logger: LoggerService,
+    events: EventsService,
+    config: KafkaEventSourceConfig,
+  ) {
+    this.kafka = new Kafka({
+      ...config.kafkaConfig,
+      logCreator: loggerServiceAdapter(logger),
+    });
+
+    this.consumers = config.kafkaConsumerConfigs.map(consumerConfig =>
+      KafkaConsumingEventPublisher.fromConfig({
+        kafkaClient: this.kafka,
+        config: consumerConfig,
+        logger,
+        events,
+      }),
+    );
+  }
+
+  async start(): Promise<void> {
+    this.consumers.map(async consumer => await consumer.start());
+  }
+
+  async shutdown(): Promise<void> {
+    this.consumers.map(async consumer => await consumer.shutdown());
+  }
+}
diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts
new file mode 100644
index 0000000000..64c53595ad
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher';
+import { Kafka } from 'kafkajs';
+import { KafkaConsumerConfig } from './config';
+import { mockServices } from '@backstage/backend-test-utils';
+
+jest.mock('kafkajs');
+
+describe('KafkaConsumingEventPublisher', () => {
+  const mockLogger = mockServices.logger.mock();
+  const mockEvents = mockServices.events.mock();
+
+  const mockConsumer = {
+    connect: jest.fn(),
+    disconnect: jest.fn(),
+    subscribe: jest.fn(),
+    run: jest.fn(),
+  };
+
+  const mockKafkaClient = {
+    consumer: jest.fn().mockReturnValue(mockConsumer),
+  } as unknown as Kafka;
+
+  const kafkaConsumerConfig: KafkaConsumerConfig = {
+    consumerConfig: {
+      groupId: 'test-group',
+    },
+    consumerSubscribeTopics: {
+      topics: ['test-topic'],
+    },
+    backstageTopic: 'backstage-topic',
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('should create an instance from config', () => {
+    const consumer = KafkaConsumingEventPublisher.fromConfig({
+      kafkaClient: mockKafkaClient,
+      config: kafkaConsumerConfig,
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    expect(consumer).toBeInstanceOf(KafkaConsumingEventPublisher);
+  });
+
+  it('should start the consumer', async () => {
+    const consumer = KafkaConsumingEventPublisher.fromConfig({
+      kafkaClient: mockKafkaClient,
+      config: kafkaConsumerConfig,
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    await consumer.start();
+
+    expect(mockConsumer.connect).toHaveBeenCalled();
+    expect(mockConsumer.subscribe).toHaveBeenCalledWith(
+      kafkaConsumerConfig.consumerSubscribeTopics,
+    );
+    expect(mockConsumer.run).toHaveBeenCalled();
+  });
+
+  it('should shutdown the consumer', async () => {
+    const consumer = KafkaConsumingEventPublisher.fromConfig({
+      kafkaClient: mockKafkaClient,
+      config: kafkaConsumerConfig,
+      events: mockEvents,
+      logger: mockLogger,
+    });
+
+    await consumer.shutdown();
+
+    expect(mockConsumer.disconnect).toHaveBeenCalled();
+  });
+});
diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts
new file mode 100644
index 0000000000..223e4e99ac
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts
@@ -0,0 +1,109 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { LoggerService } from '@backstage/backend-plugin-api';
+import { EventParams, EventsService } from '@backstage/plugin-events-node';
+import { Consumer, ConsumerSubscribeTopics, IHeaders, Kafka } from 'kafkajs';
+import { KafkaConsumerConfig } from './config';
+
+type EventMetadata = EventParams['metadata'];
+
+/**
+ *
+ * This class subscribes to Kafka topics and publishes events received to the registered subscriber.
+ * The message payload will be used as the event payload and passed to the subscribers.
+ *
+ * @public
+ */
+export class KafkaConsumingEventPublisher {
+  private readonly kafkaConsumer: Consumer;
+  private readonly consumerSubscribeTopics: ConsumerSubscribeTopics;
+  private readonly backstageTopic: string;
+  private readonly logger: LoggerService;
+
+  static fromConfig(env: {
+    kafkaClient: Kafka;
+    config: KafkaConsumerConfig;
+    events: EventsService;
+    logger: LoggerService;
+  }): KafkaConsumingEventPublisher {
+    return new KafkaConsumingEventPublisher(
+      env.kafkaClient,
+      env.logger,
+      env.events,
+      env.config,
+    );
+  }
+
+  private constructor(
+    kafkaClient: Kafka,
+    logger: LoggerService,
+    private readonly events: EventsService,
+    config: KafkaConsumerConfig,
+  ) {
+    this.kafkaConsumer = kafkaClient.consumer(config.consumerConfig);
+    this.consumerSubscribeTopics = config.consumerSubscribeTopics;
+    this.backstageTopic = config.backstageTopic;
+    const id = `events.kafka.publisher:${this.backstageTopic}`;
+    this.logger = logger.child({
+      class: KafkaConsumingEventPublisher.prototype.constructor.name,
+      groupId: config.consumerConfig.groupId,
+      kafkaTopics: config.consumerSubscribeTopics.topics.toString(),
+      backstageTopic: config.backstageTopic,
+      taskId: id,
+    });
+  }
+
+  async start(): Promise<void> {
+    try {
+      await this.kafkaConsumer.connect();
+
+      await this.kafkaConsumer.subscribe(this.consumerSubscribeTopics);
+
+      await this.kafkaConsumer.run({
+        eachMessage: async ({ message }) => {
+          this.events.publish({
+            topic: this.backstageTopic,
+            eventPayload: JSON.parse(message.value?.toString()!),
+            metadata: this.convertHeadersToMetadata(message.headers),
+          });
+        },
+      });
+    } catch (error: any) {
+      this.logger.error('Kafka consumer connection failed ', error);
+    }
+  }
+
+  async shutdown(): Promise<void> {
+    await this.kafkaConsumer.disconnect();
+  }
+
+  private convertHeadersToMetadata = (
+    headers: IHeaders | undefined,
+  ): EventParams['metadata'] => {
+    if (!headers) return undefined;
+
+    const metadata: EventMetadata = {};
+
+    Object.entries(headers).forEach(([key, value]) => {
+      // If value is an array use toString() on all values converting any Buffer types to valid strings
+      if (Array.isArray(value)) metadata[key] = value.map(v => v.toString());
+      // Always return the values using toString() to catch all Buffer types that should be converted to strings
+      else metadata[key] = value?.toString();
+    });
+
+    return metadata;
+  };
+}
diff --git a/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts
new file mode 100644
index 0000000000..b9e9379104
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { LoggerService } from '@backstage/backend-plugin-api';
+import { LogEntry, logLevel } from 'kafkajs';
+
+export const loggerServiceAdapter =
+  (loggerService: LoggerService) => (_level: logLevel) => {
+    return (entry: LogEntry) => {
+      const { namespace, level, log } = entry;
+      const { message, ...extra } = log;
+
+      const logMethods: Record<
+        logLevel,
+        (message: string, meta?: object) => void
+      > = {
+        [logLevel.ERROR]: loggerService.error,
+        [logLevel.WARN]: loggerService.warn,
+        [logLevel.INFO]: loggerService.info,
+        [logLevel.DEBUG]: loggerService.debug,
+        [logLevel.NOTHING]: () => {},
+      };
+
+      // Use loggerService method that matches the level
+      logMethods[level].call(
+        loggerService,
+        `Kafka ${namespace} ${log.message}`,
+        {
+          ...extra,
+        },
+      );
+    };
+  };
diff --git a/plugins/events-backend-module-kafka/src/publisher/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts
new file mode 100644
index 0000000000..fc7e3049b0
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts
@@ -0,0 +1,250 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { ConfigReader } from '@backstage/config';
+import { readConfig } from './config';
+
+describe('readConfig', () => {
+  it('not configured', () => {
+    const publisherConfigs = readConfig(new ConfigReader({}));
+
+    expect(publisherConfigs).toBeUndefined();
+  });
+
+  it('only required fields configured', () => {
+    const config = new ConfigReader({
+      events: {
+        modules: {
+          kafka: {
+            kafkaConsumingEventPublisher: {
+              clientId: 'backstage-events',
+              brokers: ['kafka1:9092', 'kafka2:9092'],
+              topics: [
+                {
+                  topic: 'fake1',
+                  kafka: {
+                    topics: ['topic-A'],
+                    groupId: 'my-group',
+                  },
+                },
+                {
+                  topic: 'fake2',
+                  kafka: {
+                    topics: ['topic-B'],
+                    groupId: 'my-group',
+                  },
+                },
+              ],
+            },
+          },
+        },
+      },
+    });
+
+    const publisherConfigs = readConfig(config);
+
+    expect(publisherConfigs).toBeDefined();
+
+    expect(publisherConfigs?.kafkaConsumerConfigs.length).toBe(2);
+
+    expect(publisherConfigs?.kafkaConfig.clientId).toEqual('backstage-events');
+    expect(publisherConfigs?.kafkaConfig.brokers).toEqual([
+      'kafka1:9092',
+      'kafka2:9092',
+    ]);
+    expect(publisherConfigs?.kafkaConsumerConfigs[0].backstageTopic).toEqual(
+      'fake1',
+    );
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.groupId,
+    ).toEqual('my-group');
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics,
+    ).toEqual(['topic-A']);
+  });
+
+  it('all fields configured', () => {
+    const config = new ConfigReader({
+      events: {
+        modules: {
+          kafka: {
+            kafkaConsumingEventPublisher: {
+              clientId: 'backstage-events',
+              brokers: ['kafka1:9092', 'kafka2:9092'],
+              ssl: true,
+              sasl: {
+                mechanism: 'plain',
+                username: 'username',
+                password: 'password',
+              },
+              retry: {
+                maxRetryTime: { milliseconds: 20000 },
+                initialRetryTime: { milliseconds: 200 },
+                factor: '0.4',
+                multiplier: '4',
+                retries: '10',
+              },
+              authenticationTimeout: { milliseconds: 20000 },
+              connectionTimeout: { milliseconds: 1500 },
+              requestTimeout: { milliseconds: 20000 },
+              enforceRequestTimeout: false,
+              topics: [
+                {
+                  topic: 'fake1',
+                  kafka: {
+                    topics: ['topic-A'],
+                    groupId: 'my-group',
+                    sessionTimeout: { milliseconds: 20000 },
+                    rebalanceTimeout: { milliseconds: 50000 },
+                    heartbeatInterval: { milliseconds: 2000 },
+                    metadataMaxAge: { milliseconds: 400000 },
+                    maxBytesPerPartition: 50000,
+                    minBytes: 2,
+                    maxBytes: 500000,
+                    maxWaitTime: { milliseconds: 4000 },
+                  },
+                },
+                {
+                  topic: 'fake2',
+                  kafka: {
+                    topics: ['topic-B'],
+                    groupId: 'my-group',
+                  },
+                },
+              ],
+            },
+          },
+        },
+      },
+    });
+
+    const publisherConfigs = readConfig(config);
+
+    expect(publisherConfigs).toBeDefined();
+
+    // Client configuration
+    expect(publisherConfigs?.kafkaConfig.clientId).toEqual('backstage-events');
+    expect(publisherConfigs?.kafkaConfig.brokers).toEqual([
+      'kafka1:9092',
+      'kafka2:9092',
+    ]);
+    expect(publisherConfigs?.kafkaConfig.ssl).toBeTruthy();
+    expect(publisherConfigs?.kafkaConfig.sasl).toStrictEqual({
+      mechanism: 'plain',
+      username: 'username',
+      password: 'password',
+    });
+    expect(publisherConfigs?.kafkaConfig.authenticationTimeout).toBe(20000);
+    expect(publisherConfigs?.kafkaConfig.connectionTimeout).toBe(1500);
+    expect(publisherConfigs?.kafkaConfig.requestTimeout).toBe(20000);
+    expect(publisherConfigs?.kafkaConfig.enforceRequestTimeout).toBeFalsy();
+    expect(publisherConfigs?.kafkaConfig.retry).toStrictEqual({
+      maxRetryTime: 20000,
+      initialRetryTime: 200,
+      factor: 0.4,
+      multiplier: 4,
+      retries: 10,
+    });
+
+    // Consumer configuration
+    expect(publisherConfigs?.kafkaConsumerConfigs.length).toBe(2);
+    expect(publisherConfigs?.kafkaConsumerConfigs[0].backstageTopic).toEqual(
+      'fake1',
+    );
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.groupId,
+    ).toEqual('my-group');
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics,
+    ).toEqual(['topic-A']);
+
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.sessionTimeout,
+    ).toBe(20000);
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.rebalanceTimeout,
+    ).toBe(50000);
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig
+        .heartbeatInterval,
+    ).toBe(2000);
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.metadataMaxAge,
+    ).toBe(400000);
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig
+        .maxBytesPerPartition,
+    ).toBe(50000);
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.minBytes,
+    ).toBe(2);
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.maxBytes,
+    ).toBe(500000);
+    expect(
+      publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.maxWaitTimeInMs,
+    ).toBe(4000);
+  });
+
+  it('should handle HumanDuration and string values for durations and timeouts', () => {
+    const config = new ConfigReader({
+      events: {
+        modules: {
+          kafka: {
+            kafkaConsumingEventPublisher: {
+              clientId: 'backstage-events',
+              brokers: ['kafka1:9092', 'kafka2:9092'],
+              retry: {
+                maxRetryTime: { seconds: 1 },
+                initialRetryTime: { minutes: 1 },
+                factor: 0.4,
+                multiplier: 4,
+                retries: 10,
+              },
+              authenticationTimeout: { hours: 1 },
+              connectionTimeout: { days: 1 },
+              topics: [],
+              requestTimeout: '1m',
+            },
+          },
+        },
+      },
+    });
+
+    const publisherConfigs = readConfig(config);
+
+    expect(publisherConfigs).toBeDefined();
+
+    // Client configuration
+    expect(publisherConfigs?.kafkaConfig.clientId).toEqual('backstage-events');
+    expect(publisherConfigs?.kafkaConfig.brokers).toEqual([
+      'kafka1:9092',
+      'kafka2:9092',
+    ]);
+    expect(publisherConfigs?.kafkaConfig.authenticationTimeout).toBe(3600000);
+    expect(publisherConfigs?.kafkaConfig.connectionTimeout).toBe(86400000);
+    expect(publisherConfigs?.kafkaConfig.requestTimeout).toBe(60000);
+    expect(publisherConfigs?.kafkaConfig.retry).toStrictEqual({
+      maxRetryTime: 1000,
+      initialRetryTime: 60000,
+      factor: 0.4,
+      multiplier: 4,
+      retries: 10,
+    });
+
+    // Consumer configuration
+    expect(publisherConfigs?.kafkaConsumerConfigs.length).toBe(0);
+  });
+});
diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts
new file mode 100644
index 0000000000..aa5921a7f3
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/config.ts
@@ -0,0 +1,159 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { Config, readDurationFromConfig } from '@backstage/config';
+import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs';
+import { Duration } from 'luxon';
+
+/**
+ * @public
+ */
+export interface KafkaConsumerConfig {
+  backstageTopic: string;
+  consumerConfig: ConsumerConfig;
+  consumerSubscribeTopics: ConsumerSubscribeTopics;
+}
+
+/**
+ * @public
+ */
+export interface KafkaEventSourceConfig {
+  kafkaConfig: KafkaConfig;
+  kafkaConsumerConfigs: KafkaConsumerConfig[];
+}
+
+const CONFIG_PREFIX_PUBLISHER =
+  'events.modules.kafka.kafkaConsumingEventPublisher';
+
+/**
+ * Reads an optional HumanDuration from the config and returns the value in milliseconds if the key is defined.
+ *
+ * @param config - The configuration object to read from.
+ * @param key - The key to look up in the configuration.
+ * @returns The duration in milliseconds, or undefined if the key is not defined.
+ */
+const readOptionalHumanDurationInMs = (
+  config: Config,
+  key: string,
+): number | undefined => {
+  const humanDuration = config.has(key)
+    ? readDurationFromConfig(config, { key })
+    : undefined;
+
+  if (!humanDuration) return undefined;
+
+  return Duration.fromObject(humanDuration).as('milliseconds');
+};
+
+export const readConfig = (
+  config: Config,
+): KafkaEventSourceConfig | undefined => {
+  const kafkaConfig = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER);
+
+  if (!kafkaConfig) {
+    return undefined;
+  }
+
+  const clientId = kafkaConfig.getString('clientId');
+  const brokers = kafkaConfig.getStringArray('brokers');
+
+  const authenticationTimeout = readOptionalHumanDurationInMs(
+    kafkaConfig,
+    'authenticationTimeout',
+  );
+
+  const connectionTimeout = readOptionalHumanDurationInMs(
+    kafkaConfig,
+    'connectionTimeout',
+  );
+  const requestTimeout = readOptionalHumanDurationInMs(
+    kafkaConfig,
+    'requestTimeout',
+  );
+  const enforceRequestTimeout = kafkaConfig.getOptionalBoolean(
+    'enforceRequestTimeout',
+  );
+
+  const ssl = kafkaConfig.getOptional('ssl') as KafkaConfig['ssl'];
+  const sasl = kafkaConfig.getOptional('sasl') as KafkaConfig['sasl'];
+
+  const retry: KafkaConfig['retry'] = {
+    maxRetryTime: readOptionalHumanDurationInMs(
+      kafkaConfig,
+      'retry.maxRetryTime',
+    ),
+    initialRetryTime: readOptionalHumanDurationInMs(
+      kafkaConfig,
+      'retry.initialRetryTime',
+    ),
+    factor: kafkaConfig.getOptionalNumber('retry.factor'),
+    multiplier: kafkaConfig.getOptionalNumber('retry.multiplier'),
+    retries: kafkaConfig.getOptionalNumber('retry.retries'),
+  };
+
+  const kafkaConsumerConfigs: KafkaConsumerConfig[] = kafkaConfig
+    .getConfigArray('topics')
+    .map(topic => {
+      return {
+        backstageTopic: topic.getString('topic'),
+        consumerConfig: {
+          groupId: topic.getString('kafka.groupId'),
+          sessionTimeout: readOptionalHumanDurationInMs(
+            topic,
+            'kafka.sessionTimeout',
+          ),
+          rebalanceTimeout: readOptionalHumanDurationInMs(
+            topic,
+            'kafka.rebalanceTimeout',
+          ),
+          heartbeatInterval: readOptionalHumanDurationInMs(
+            topic,
+            'kafka.heartbeatInterval',
+          ),
+          metadataMaxAge: readOptionalHumanDurationInMs(
+            topic,
+            'kafka.metadataMaxAge',
+          ),
+          maxBytesPerPartition: topic.getOptionalNumber(
+            'kafka.maxBytesPerPartition',
+          ),
+          minBytes: topic.getOptionalNumber('kafka.minBytes'),
+          maxBytes: topic.getOptionalNumber('kafka.maxBytes'),
+          maxWaitTimeInMs: readOptionalHumanDurationInMs(
+            topic,
+            'kafka.maxWaitTime',
+          ),
+        },
+        consumerSubscribeTopics: {
+          topics: topic.getStringArray('kafka.topics'),
+        },
+      };
+    });
+
+  return {
+    kafkaConfig: {
+      clientId,
+      brokers,
+      ssl,
+      sasl,
+      authenticationTimeout,
+      connectionTimeout,
+      requestTimeout,
+      enforceRequestTimeout,
+      retry,
+    },
+    kafkaConsumerConfigs,
+  };
+};
diff --git a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts
new file mode 100644
index 0000000000..816793f635
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import { createServiceFactory } from '@backstage/backend-plugin-api';
+import { mockServices, startTestBackend } from '@backstage/backend-test-utils';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
+import { TestEventsService } from '@backstage/plugin-events-backend-test-utils';
+import { eventsModuleKafkaConsumingEventPublisher } from './eventsModuleKafkaConsumingEventPublisher';
+import { KafkaConsumerClient } from '../publisher/KafkaConsumerClient';
+
+jest.mock('../publisher/KafkaConsumerClient');
+
+describe('eventsModuleKafkaConsumingEventPublisher', () => {
+  it('should be correctly wired and set up', async () => {
+    const events = new TestEventsService();
+    const eventsServiceFactory = createServiceFactory({
+      service: eventsServiceRef,
+      deps: {},
+      async factory({}) {
+        return events;
+      },
+    });
+
+    const mockKafkaConsumerClient = {
+      start: jest.fn(),
+      shutdown: jest.fn(),
+    };
+    (KafkaConsumerClient.fromConfig as jest.Mock).mockReturnValue(
+      mockKafkaConsumerClient,
+    );
+
+    await startTestBackend({
+      features: [
+        eventsServiceFactory,
+        eventsModuleKafkaConsumingEventPublisher,
+        mockServices.rootConfig.factory({
+          data: {
+            events: {
+              modules: {
+                kafka: {
+                  kafkaConsumingEventPublisher: {
+                    clientId: 'backstage-events',
+                    brokers: ['kafka1:9092', 'kafka2:9092'],
+                    topics: {
+                      fake1: {
+                        kafka: {
+                          topics: ['topic-A'],
+                          groupId: 'my-group',
+                        },
+                      },
+                      fake2: {
+                        kafka: {
+                          topics: ['topic-B'],
+                          groupId: 'my-group',
+                        },
+                      },
+                    },
+                  },
+                },
+              },
+            },
+          },
+        }),
+      ],
+    });
+
+    // Verify that the Kafka consumer client was started
+    expect(mockKafkaConsumerClient.start).toHaveBeenCalled();
+
+    // Verify that the shutdown hook was registered
+    expect(mockKafkaConsumerClient.shutdown).not.toHaveBeenCalled();
+  });
+});
diff --git a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts
new file mode 100644
index 0000000000..b524454ca0
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+import {
+  coreServices,
+  createBackendModule,
+} from '@backstage/backend-plugin-api';
+import { KafkaConsumerClient } from '../publisher/KafkaConsumerClient';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
+
+/**
+ * Kafka module for the Events plugin.
+ *
+ * @public
+ */
+export const eventsModuleKafkaConsumingEventPublisher = createBackendModule({
+  pluginId: 'events',
+  moduleId: 'kafka-consuming-event-publisher',
+  register(env) {
+    env.registerInit({
+      deps: {
+        config: coreServices.rootConfig,
+        events: eventsServiceRef,
+        logger: coreServices.logger,
+        lifecycle: coreServices.lifecycle,
+      },
+      async init({ config, logger, events, lifecycle }) {
+        const kafka = KafkaConsumerClient.fromConfig({
+          config,
+          events,
+          logger,
+        });
+
+        if (!kafka) {
+          return;
+        }
+
+        await kafka.start();
+
+        lifecycle.addShutdownHook(async () => await kafka.shutdown());
+      },
+    });
+  },
+});
diff --git a/yarn.lock b/yarn.lock
index b6d6d1e276..c417643c4d 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -6700,6 +6700,22 @@ __metadata:
   languageName: unknown
   linkType: soft
 
+"@backstage/plugin-events-backend-module-kafka@workspace:plugins/events-backend-module-kafka":
+  version: 0.0.0-use.local
+  resolution: "@backstage/plugin-events-backend-module-kafka@workspace:plugins/events-backend-module-kafka"
+  dependencies:
+    "@backstage/backend-plugin-api": "workspace:^"
+    "@backstage/backend-test-utils": "workspace:^"
+    "@backstage/cli": "workspace:^"
+    "@backstage/config": "workspace:^"
+    "@backstage/plugin-events-backend-test-utils": "workspace:^"
+    "@backstage/plugin-events-node": "workspace:^"
+    "@backstage/types": "workspace:^"
+    kafkajs: "npm:^2.2.4"
+    luxon: "npm:^3.0.0"
+  languageName: unknown
+  linkType: soft
+
 "@backstage/plugin-events-backend-test-utils@workspace:^, @backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils":
   version: 0.0.0-use.local
   resolution: "@backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils"
@@ -35319,6 +35335,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"kafkajs@npm:^2.2.4":
+  version: 2.2.4
+  resolution: "kafkajs@npm:2.2.4"
+  checksum: 10/75eb0d221397085f90e51f8a2d752495c9fa9c1b3a1a6db610cd7074fa8c52777f295832fd0a7c49cded5e574337a09fafa8c3f7cf1caa38f4dc9aa20fcfb7df
+  languageName: node
+  linkType: hard
+
 "keygrip@npm:~1.1.0":
   version: 1.1.0
   resolution: "keygrip@npm:1.1.0"