diff --git a/README.md b/README.md
index f63446a..69ffeaf 100644
--- a/README.md
+++ b/README.md
@@ -11,6 +11,9 @@ Databricks input plugin for Embulk loads records from Databricks.
 
 - **driver_path**: path to the jar file of the JDBC driver. If not set, [the bundled JDBC driver](https://docs.databricks.com/en/integrations/jdbc/index.html) will be used. (string, optional)
 - **options**: extra JDBC properties (hash, default: {})
+- **user_agent**: set user agent property to JDBC connection. If 'UserAgentEntry' property is specified in the **options**, it will be overwritten by this value. (hash, optional)
+  - **product_name**: product name of user agent (string, default: "unknown")
+  - **product_version**: product version of user agent (string, default: "0.0.0")
 - **server_hostname**: The Databricks compute resource’s Server Hostname value, see [Compute settings for the Databricks JDBC Driver](https://docs.databricks.com/en/integrations/jdbc/compute.html). (string, required)
 - **http_path**: The Databricks compute resource’s HTTP Path value, see [Compute settings for the Databricks JDBC Driver](https://docs.databricks.com/en/integrations/jdbc/compute.html). (string, required)
 - **personal_access_token**: The Databaricks personal_access_token, see [Authentication settings for the Databricks JDBC Driver](https://docs.databricks.com/en/integrations/jdbc/authentication.html#authentication-pat). (string, required)
diff --git a/src/main/java/org/embulk/input/DatabricksInputPlugin.java b/src/main/java/org/embulk/input/DatabricksInputPlugin.java
old mode 100644
new mode 100755
index 0511ac3..9d5de09
--- a/src/main/java/org/embulk/input/DatabricksInputPlugin.java
+++ b/src/main/java/org/embulk/input/DatabricksInputPlugin.java
@@ -11,6 +11,7 @@
 import org.embulk.spi.Schema;
 import org.embulk.util.config.Config;
 import org.embulk.util.config.ConfigDefault;
+import org.embulk.util.config.Task;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,6 +40,20 @@ public interface DatabricksPluginTask extends PluginTask {
     @Config("schema_name")
     @ConfigDefault("null")
     public Optional<String> getSchemaName();
+
+    @Config("user_agent")
+    @ConfigDefault("{}")
+    public UserAgentEntry getUserAgentEntry();
+
+    public interface UserAgentEntry extends Task {
+      @Config("product_name")
+      @ConfigDefault("\"unknown\"")
+      public String getProductName();
+
+      @Config("product_version")
+      @ConfigDefault("\"0.0.0\"")
+      public String getProductVersion();
+    }
   }
 
   @Override
@@ -76,6 +91,11 @@ protected JdbcInputConnection newConnection(PluginTask task) throws SQLException
     }
     props.putAll(t.getOptions());
 
+    // overwrite UserAgentEntry property if the same property is set in options
+    String productName = t.getUserAgentEntry().getProductName();
+    String productVersion = t.getUserAgentEntry().getProductVersion();
+    props.put("UserAgentEntry", productName + "/" + productVersion);
+
     logConnectionProperties(url, props);
     Connection c = DriverManager.getConnection(url, props);
     return new DatabricksInputConnection(