diff --git a/src/main/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryService.java b/src/main/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryService.java
index 11d1054ef..2ad46e4f9 100644
--- a/src/main/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryService.java
+++ b/src/main/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryService.java
@@ -40,6 +40,8 @@ public abstract class SnowflakeTelemetryService {
   private static final String KAFKA_VERSION = "kafka_version";
   protected static final String IS_PIPE_CLOSING = "is_pipe_closing";
   protected static final String IS_CHANNEL_CLOSING = "is_channel_closing";
+  public static final String JDK_VERSION = "jdk_version";
+  public static final String JDK_DISTRIBUTION = "jdk_distribution";
 
   // Telemetry instance fetched from JDBC
   protected Telemetry telemetry;
@@ -77,8 +79,13 @@ public void reportKafkaConnectStart(
       final long startTime, final Map<String, String> userProvidedConfig) {
     ObjectNode dataObjectNode = getObjectNode();
 
+    String jdkVersion = System.getProperty("java.version");
+    String jdkDistribution = System.getProperty("java.vendor");
+
     dataObjectNode.put(START_TIME, startTime);
     dataObjectNode.put(KAFKA_VERSION, AppInfoParser.getVersion());
+    dataObjectNode.put(JDK_VERSION, jdkVersion);
+    dataObjectNode.put(JDK_DISTRIBUTION, jdkDistribution);
     addUserConnectorPropertiesToDataNode(userProvidedConfig, dataObjectNode);
 
     send(SnowflakeTelemetryService.TelemetryType.KAFKA_START, dataObjectNode);
diff --git a/src/test/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryServiceTest.java b/src/test/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryServiceTest.java
index 45d0764ad..9142ca9a8 100644
--- a/src/test/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryServiceTest.java
+++ b/src/test/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryServiceTest.java
@@ -111,6 +111,9 @@ public void testReportKafkaConnectStart(IngestionMethodConfig ingestionMethodCon
         dataNode.get(TelemetryConstants.START_TIME).asLong() <= System.currentTimeMillis()
             && dataNode.get(TelemetryConstants.START_TIME).asLong() >= this.startTime);
 
+    assertNotNull(dataNode.get("jdk_version"));
+    assertNotNull(dataNode.get("jdk_distribution"));
+
     validateBufferProperties(dataNode);
     validateKeyAndValueConverter(dataNode);
   }