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..0bdbd5cf1e
--- /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](path/to/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..4c0d06c77e
--- /dev/null
+++ b/plugins/events-backend-module-kafka/config.d.ts
@@ -0,0 +1,202 @@
+/*
+ * 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.
+ */
+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 in milliseconds
+             * Default: 30000 ms.
+             */
+            maxRetryTime: number;
+
+            /**
+             * (Optional) Initial value used to calculate the retry in milliseconds (This is still randomized following the randomization factor)
+             * Default: 300 ms.
+             */
+            initialRetryTime: number;
+
+            /**
+             * (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 in ms for authentication requests.
+           * Default: 10000 ms.
+           */
+          authenticationTimeout: number;
+
+          /**
+           * (Optional) Time in milliseconds to wait for a successful connection.
+           * Default: 1000 ms.
+           */
+          connectionTimeout: number;
+
+          /**
+           * (Optional) Time in milliseconds to wait for a successful request.
+           * Default: 30000 ms.
+           */
+          requestTimeout: number;
+
+          /**
+           * (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 in milliseconds 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: number;
+
+              /**
+               * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group
+               * Default: 60000 ms.
+               */
+              rebalanceTimeout: number;
+
+              /**
+               * (Optional) The expected time in milliseconds 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: number;
+
+              /**
+               * (Optional) The period of time in milliseconds 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: number;
+
+              /**
+               * (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 maxWaitTimeInMs 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 in milliseconds 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
+               */
+              maxWaitTimeInMs: number;
+            };
+          }>;
+        };
+      };
+    };
+  };
+}
diff --git a/plugins/events-backend-module-kafka/package.json b/plugins/events-backend-module-kafka/package.json
new file mode 100644
index 0000000000..1750caf36b
--- /dev/null
+++ b/plugins/events-backend-module-kafka/package.json
@@ -0,0 +1,41 @@
+{
+  "name": "@backstage/plugin-events-backend-module-kafka",
+  "description": "The kafka backend module for the events plugin.",
+  "version": "0.0.0",
+  "main": "src/index.ts",
+  "types": "src/index.ts",
+  "license": "Apache-2.0",
+  "private": true,
+  "publishConfig": {
+    "access": "public",
+    "main": "dist/index.cjs.js",
+    "types": "dist/index.d.ts"
+  },
+  "backstage": {
+    "role": "backend-plugin-module"
+  },
+  "scripts": {
+    "start": "backstage-cli package start",
+    "build": "backstage-cli package build",
+    "lint": "backstage-cli package lint",
+    "test": "backstage-cli package test",
+    "clean": "backstage-cli package clean",
+    "prepack": "backstage-cli package prepack",
+    "postpack": "backstage-cli package postpack"
+  },
+  "dependencies": {
+    "@backstage/backend-plugin-api": "workspace:^",
+    "@backstage/config": "workspace:^",
+    "@backstage/plugin-events-node": "workspace:^",
+    "@backstage/types": "workspace:^",
+    "kafkajs": "^2.2.4"
+  },
+  "devDependencies": {
+    "@backstage/backend-test-utils": "workspace:^",
+    "@backstage/cli": "workspace:^",
+    "@backstage/plugin-events-backend-test-utils": "workspace:^"
+  },
+  "files": [
+    "dist"
+  ]
+}
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..b8fd142d0e
--- /dev/null
+++ b/plugins/events-backend-module-kafka/report.api.md
@@ -0,0 +1,50 @@
+## API Report File for "@internal/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';
+import { Config } from '@backstage/config';
+import { ConsumerConfig } from 'kafkajs';
+import { ConsumerSubscribeTopics } from 'kafkajs';
+import { EventsService } from '@backstage/plugin-events-node';
+import { Kafka } from 'kafkajs';
+import { LoggerService } from '@backstage/backend-plugin-api';
+
+// @public
+const eventsModuleKafkaConsumingEventPublisher: BackendFeature;
+export default eventsModuleKafkaConsumingEventPublisher;
+
+// @public (undocumented)
+export class KafkaConsumerClient {
+  // (undocumented)
+  static fromConfig(env: {
+    config: Config;
+    events: EventsService;
+    logger: LoggerService;
+  }): KafkaConsumerClient;
+  // (undocumented)
+  shutdown(): Promise<void>;
+  // (undocumented)
+  start(): Promise<void>;
+}
+
+// @public (undocumented)
+export class KafkaConsumingEventPublisher {
+  // (undocumented)
+  static fromConfig(env: {
+    kafkaClient: Kafka;
+    config: KafkaConsumerConfig;
+    events: EventsService;
+    logger: LoggerService;
+  }): KafkaConsumingEventPublisher;
+  // (undocumented)
+  shutdown(): Promise<void>;
+  // (undocumented)
+  start(): Promise<void>;
+}
+
+// Warnings were encountered during analysis:
+//
+// src/publisher/KafkaConsumingEventPublisher.d.ts:13:9 - (ae-forgotten-export) The symbol "KafkaConsumerConfig" needs to be exported by the entry point index.d.ts
+```
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..ff4fe2a209
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/index.ts
@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+
+export { KafkaConsumerClient } from './publisher/KafkaConsumerClient';
+export { KafkaConsumingEventPublisher } from './publisher/KafkaConsumingEventPublisher';
+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..c0d6809514
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts
@@ -0,0 +1,118 @@
+/*
+ * 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 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,
+    });
+
+    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,
+    });
+
+    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..06a7b2594c
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts
@@ -0,0 +1,67 @@
+/*
+ * 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 { LoggerServiceCreator } from './LoggerServiceCreator';
+
+export class KafkaConsumerClient {
+  private readonly kafka: Kafka;
+  private readonly consumers: KafkaConsumingEventPublisher[];
+
+  static fromConfig(env: {
+    config: Config;
+    events: EventsService;
+    logger: LoggerService;
+  }): KafkaConsumerClient {
+    return new KafkaConsumerClient(
+      env.logger,
+      env.events,
+      readConfig(env.config),
+    );
+  }
+
+  private constructor(
+    logger: LoggerService,
+    events: EventsService,
+    config: KafkaEventSourceConfig,
+  ) {
+    this.kafka = new Kafka({
+      ...config.kafkaConfig,
+      logCreator: LoggerServiceCreator(logger),
+    });
+
+    this.consumers = config.kafkaConsumerConfig.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..c0bf94d3c8
--- /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',
+    },
+    consumerSubscribeConfig: {
+      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.consumerSubscribeConfig,
+    );
+    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..90db289a8a
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts
@@ -0,0 +1,104 @@
+/*
+ * 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'];
+
+export class KafkaConsumingEventPublisher {
+  private readonly kafkaConsumer: Consumer;
+  private readonly consumerSubscribeOptions: 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.consumerSubscribeOptions = config.consumerSubscribeConfig;
+    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.consumerSubscribeConfig.topics.toString(),
+      backstageTopic: config.backstageTopic,
+      taskId: id,
+    });
+  }
+
+  async start(): Promise<void> {
+    try {
+      await this.kafkaConsumer.connect();
+      this.logger.info('Kafka consumer connected');
+
+      await this.kafkaConsumer.subscribe(this.consumerSubscribeOptions);
+
+      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();
+    this.logger.info('Kafka consumer disconnected');
+  }
+
+  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/LoggerServiceCreator.ts b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts
new file mode 100644
index 0000000000..0dd5f9cd84
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.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 LoggerServiceCreator =
+  (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..88c1d9747c
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts
@@ -0,0 +1,197 @@
+/*
+ * 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 config = new ConfigReader({});
+
+    expect(() => {
+      readConfig(config);
+    }).toThrow();
+  });
+
+  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.kafkaConsumerConfig.length).toBe(2);
+
+    expect(publisherConfigs.kafkaConfig.clientId).toEqual('backstage-events');
+    expect(publisherConfigs.kafkaConfig.brokers).toEqual([
+      'kafka1:9092',
+      'kafka2:9092',
+    ]);
+    expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual(
+      'fake1',
+    );
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId,
+    ).toEqual('my-group');
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.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: '20000',
+                initialRetryTime: '200',
+                factor: '0.4',
+                multiplier: '4',
+                retries: '10',
+              },
+              authenticationTimeout: 20000,
+              connectionTimeout: 1500,
+              requestTimeout: 20000,
+              enforceRequestTimeout: false,
+              topics: [
+                {
+                  topic: 'fake1',
+                  kafka: {
+                    topics: ['topic-A'],
+                    groupId: 'my-group',
+                    sessionTimeout: 20000,
+                    rebalanceTimeout: 50000,
+                    heartbeatInterval: 2000,
+                    metadataMaxAge: 400000,
+                    maxBytesPerPartition: 50000,
+                    minBytes: 2,
+                    maxBytes: 500000,
+                    maxWaitTimeInMs: 4000,
+                  },
+                },
+                {
+                  topic: 'fake2',
+                  kafka: {
+                    topics: ['topic-B'],
+                    groupId: 'my-group',
+                  },
+                },
+              ],
+            },
+          },
+        },
+      },
+    });
+
+    const publisherConfigs = readConfig(config);
+
+    // 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.kafkaConsumerConfig.length).toBe(2);
+    expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual(
+      'fake1',
+    );
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId,
+    ).toEqual('my-group');
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.topics,
+    ).toEqual(['topic-A']);
+
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.sessionTimeout,
+    ).toBe(20000);
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.rebalanceTimeout,
+    ).toBe(50000);
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.heartbeatInterval,
+    ).toBe(2000);
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.metadataMaxAge,
+    ).toBe(400000);
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig
+        .maxBytesPerPartition,
+    ).toBe(50000);
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.minBytes,
+    ).toBe(2);
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxBytes,
+    ).toBe(500000);
+    expect(
+      publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxWaitTimeInMs,
+    ).toBe(4000);
+  });
+});
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..206ea0b8ac
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/publisher/config.ts
@@ -0,0 +1,90 @@
+/*
+ * 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 } from '@backstage/config';
+import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs';
+
+export interface KafkaConsumerConfig {
+  backstageTopic: string;
+  consumerConfig: ConsumerConfig;
+  consumerSubscribeConfig: ConsumerSubscribeTopics;
+}
+
+export interface KafkaEventSourceConfig {
+  kafkaConfig: KafkaConfig;
+  kafkaConsumerConfig: KafkaConsumerConfig[];
+}
+
+const CONFIG_PREFIX_PUBLISHER =
+  'events.modules.kafka.kafkaConsumingEventPublisher';
+
+export const readConfig = (config: Config): KafkaEventSourceConfig => {
+  const kafkaConfig = config.getConfig(CONFIG_PREFIX_PUBLISHER);
+
+  const clientId = kafkaConfig.getString('clientId');
+  const brokers = kafkaConfig.getStringArray('brokers');
+
+  const authenticationTimeout = kafkaConfig.getOptionalNumber(
+    'authenticationTimeout',
+  );
+  const connectionTimeout = kafkaConfig.getOptionalNumber('connectionTimeout');
+  const requestTimeout = kafkaConfig.getOptionalNumber('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.getOptional('retry') as KafkaConfig['retry'];
+
+  const kafkaConsumerConfig: KafkaConsumerConfig[] = kafkaConfig
+    .getConfigArray('topics')
+    .map(topic => {
+      return {
+        backstageTopic: topic.getString('topic'),
+        consumerConfig: {
+          groupId: topic.getString('kafka.groupId'),
+          sessionTimeout: topic.getOptionalNumber('kafka.sessionTimeout'),
+          rebalanceTimeout: topic.getOptionalNumber('kafka.rebalanceTimeout'),
+          heartbeatInterval: topic.getOptionalNumber('kafka.heartbeatInterval'),
+          metadataMaxAge: topic.getOptionalNumber('kafka.metadataMaxAge'),
+          maxBytesPerPartition: topic.getOptionalNumber(
+            'kafka.maxBytesPerPartition',
+          ),
+          minBytes: topic.getOptionalNumber('kafka.minBytes'),
+          maxBytes: topic.getOptionalNumber('kafka.maxBytes'),
+          maxWaitTimeInMs: topic.getOptionalNumber('kafka.maxWaitTimeInMs'),
+        },
+        consumerSubscribeConfig: {
+          topics: topic.getStringArray('kafka.topics'),
+        },
+      };
+    });
+
+  return {
+    kafkaConfig: {
+      clientId,
+      brokers,
+      ssl,
+      sasl,
+      authenticationTimeout,
+      connectionTimeout,
+      requestTimeout,
+      enforceRequestTimeout,
+      retry,
+    },
+    kafkaConsumerConfig,
+  };
+};
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..a458a39662
--- /dev/null
+++ b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts
@@ -0,0 +1,52 @@
+/*
+ * 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,
+        });
+
+        await kafka.start();
+
+        lifecycle.addShutdownHook(async () => await kafka.shutdown());
+      },
+    });
+  },
+});
diff --git a/yarn.lock b/yarn.lock
index 1e822bfc7c..b1dea974cb 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -35254,6 +35254,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"kafkajs@npm:^2.2.4":
+  version: 2.2.4
+  resolution: "kafkajs@npm:2.2.4"
+  checksum: 83e9e8bc50a09b142f4ff79f6a2bd88ecc21b83bcefe6621ab1716118d624886befb7371731274f67812ce35dd50b53140ff3b49a06e5d9169fe6b164d72fea5
+  languageName: node
+  linkType: hard
+
 "keygrip@npm:~1.1.0":
   version: 1.1.0
   resolution: "keygrip@npm:1.1.0"