Bug: [bug]: fix timestamp sent to kafka events



rdkafka library expects time in millis but we send seconds.
This can be reflected in the kafka-ui at localhost:8090 as well.


We can fix this by either truncating the unix_timestamp_nanos or padding the seconds.
diff --git a/crates/router/src/services/kafka.rs b/crates/router/src/services/kafka.rs
index 58657c59fd8..6f2aa9fcd10 100644
--- a/crates/router/src/services/kafka.rs
+++ b/crates/router/src/services/kafka.rs
@@ -259,7 +259,7 @@ impl KafkaProducer {
                     .timestamp(
                         event
                             .creation_timestamp()
-                            .unwrap_or_else(|| OffsetDateTime::now_utc().unix_timestamp()),
+                            .unwrap_or_else(|| OffsetDateTime::now_utc().unix_timestamp() * 1_000),
                     ),
             )
             .map_err(|(error, record)| report!(error).attach_printable(format!("{record:?}")))
@@ -476,7 +476,7 @@ impl MessagingInterface for KafkaProducer {
                     .key(&data.identifier())
                     .payload(&json_data)
                     .timestamp(
-                        (timestamp.assume_utc().unix_timestamp_nanos() / 1_000)
+                        (timestamp.assume_utc().unix_timestamp_nanos() / 1_000_000)
                             .to_i64()
                             .unwrap_or_else(|| {
                                 // kafka producer accepts milliseconds
diff --git a/crates/router/src/services/kafka/dispute.rs b/crates/router/src/services/kafka/dispute.rs
index c01089855c6..1950fb253a7 100644
--- a/crates/router/src/services/kafka/dispute.rs
+++ b/crates/router/src/services/kafka/dispute.rs
@@ -70,10 +70,6 @@ impl<'a> super::KafkaMessage for KafkaDispute<'a> {
         )
     }
 
-    fn creation_timestamp(&self) -> Option<i64> {
-        Some(self.modified_at.unix_timestamp())
-    }
-
     fn event_type(&self) -> crate::events::EventType {
         crate::events::EventType::Dispute
     }
diff --git a/crates/router/src/services/kafka/payment_attempt.rs b/crates/router/src/services/kafka/payment_attempt.rs
index dfa7da0d0a4..86ea378bfa3 100644
--- a/crates/router/src/services/kafka/payment_attempt.rs
+++ b/crates/router/src/services/kafka/payment_attempt.rs
@@ -108,10 +108,6 @@ impl<'a> super::KafkaMessage for KafkaPaymentAttempt<'a> {
         )
     }
 
-    fn creation_timestamp(&self) -> Option<i64> {
-        Some(self.modified_at.unix_timestamp())
-    }
-
     fn event_type(&self) -> crate::events::EventType {
         crate::events::EventType::PaymentAttempt
     }
diff --git a/crates/router/src/services/kafka/payment_intent.rs b/crates/router/src/services/kafka/payment_intent.rs
index ad61e102955..bcdf23e7941 100644
--- a/crates/router/src/services/kafka/payment_intent.rs
+++ b/crates/router/src/services/kafka/payment_intent.rs
@@ -66,10 +66,6 @@ impl<'a> super::KafkaMessage for KafkaPaymentIntent<'a> {
         format!("{}_{}", self.merchant_id, self.payment_id)
     }
 
-    fn creation_timestamp(&self) -> Option<i64> {
-        Some(self.modified_at.unix_timestamp())
-    }
-
     fn event_type(&self) -> crate::events::EventType {
         crate::events::EventType::PaymentIntent
     }
diff --git a/crates/router/src/services/kafka/payout.rs b/crates/router/src/services/kafka/payout.rs
index 6b25c724eb8..a07b90ee288 100644
--- a/crates/router/src/services/kafka/payout.rs
+++ b/crates/router/src/services/kafka/payout.rs
@@ -80,10 +80,6 @@ impl<'a> super::KafkaMessage for KafkaPayout<'a> {
         format!("{}_{}", self.merchant_id, self.payout_attempt_id)
     }
 
-    fn creation_timestamp(&self) -> Option<i64> {
-        Some(self.last_modified_at.unix_timestamp())
-    }
-
     fn event_type(&self) -> crate::events::EventType {
         crate::events::EventType::Payout
     }

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [x] Bugfix
- [ ] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
 fixes #4706

Convert secs to millis when sending creation timestamp to kafka

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
- This will fix problems encountered down the pipeline with data retention

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
- the timestamp listed in kafka-ui
- can't be tested without infra access and no visible changes on UI

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [ ] I formatted the code `cargo +nightly fmt --all`
- [ ] I addressed lints thrown by `cargo clippy`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
