diff --git a/v1_examples/destination/java/Dockerfile b/v1_examples/destination/java/Dockerfile
index 7f5ec60..d6b64e2 100644
--- a/v1_examples/destination/java/Dockerfile
+++ b/v1_examples/destination/java/Dockerfile
@@ -1,6 +1,6 @@
 FROM openjdk:11-jre-slim
 
-EXPOSE 50051
+EXPOSE 50052
 
 RUN mkdir /app
 
diff --git a/v2_examples/destination/java/.gitattributes b/v2_examples/destination/java/.gitattributes
new file mode 100644
index 0000000..00a51af
--- /dev/null
+++ b/v2_examples/destination/java/.gitattributes
@@ -0,0 +1,6 @@
+#
+# https://help.github.com/articles/dealing-with-line-endings/
+#
+# These are explicitly windows files and should use crlf
+*.bat           text eol=crlf
+
diff --git a/v2_examples/destination/java/Dockerfile b/v2_examples/destination/java/Dockerfile
new file mode 100644
index 0000000..d6b64e2
--- /dev/null
+++ b/v2_examples/destination/java/Dockerfile
@@ -0,0 +1,9 @@
+FROM openjdk:11-jre-slim
+
+EXPOSE 50052
+
+RUN mkdir /app
+
+COPY build/libs/*.jar /app/JavaDestination.jar
+
+ENTRYPOINT ["java", "-jar", "--illegal-access=debug", "--add-opens=java.base/java.io=ALL-UNNAMED", "--add-opens=java.base/java.lang=ALL-UNNAMED", "--add-opens=java.base/java.nio.charset=ALL-UNNAMED", "--add-opens=java.base/java.util=ALL-UNNAMED", "/app/JavaDestination.jar"]
diff --git a/v2_examples/destination/java/README.md b/v2_examples/destination/java/README.md
new file mode 100644
index 0000000..cf8e93e
--- /dev/null
+++ b/v2_examples/destination/java/README.md
@@ -0,0 +1,19 @@
+# Java Destination Example
+
+## Pre-requisites
+- JDK v17
+- Gradle 8
+
+## Steps
+1. Copy proto files from the root folder
+```
+> gradle copyProtos
+```
+2. Build the Jar
+```
+> gradle jar
+```
+3. Run the Jar
+```
+> java -jar build/libs/JavaDestination.jar 
+```
diff --git a/v2_examples/destination/java/build.gradle b/v2_examples/destination/java/build.gradle
new file mode 100644
index 0000000..1474354
--- /dev/null
+++ b/v2_examples/destination/java/build.gradle
@@ -0,0 +1,85 @@
+plugins {
+    // Provide convenience executables for trying out the examples.
+    id 'application'
+    // ASSUMES GRADLE 5.6 OR HIGHER. Use plugin version 0.8.10 with earlier gradle versions
+    id 'com.google.protobuf' version '0.9.1'
+    // Generate IntelliJ IDEA's .idea & .iml project files
+    id 'idea'
+    id 'java'
+}
+
+repositories { 
+    maven { // The google mirror is less flaky than mavenCentral()
+        url "https://maven-central.storage-download.googleapis.com/maven2/"
+
+        artifactUrls "https://repo1.maven.org/maven2/com/fasterxml/jackson/core/jackson-core/"
+    }
+    mavenCentral()
+}
+
+sourceCompatibility = 1.8
+targetCompatibility = 1.8
+
+def grpcVersion = '1.61.1'
+def protobufVersion = '3.25.2'
+def protocVersion = protobufVersion
+
+dependencies {
+    implementation "io.grpc:grpc-protobuf:${grpcVersion}"
+    implementation "io.grpc:grpc-stub:${grpcVersion}"
+    compileOnly "org.apache.tomcat:annotations-api:6.0.53"
+
+    implementation "com.google.protobuf:protobuf-java-util:${protobufVersion}"
+
+    runtimeOnly "io.grpc:grpc-netty-shaded:${grpcVersion}"
+
+    implementation 'com.fasterxml.jackson.core:jackson-databind:2.15.2'
+    implementation "com.fasterxml.jackson.core:jackson-core:2.15.2"
+
+    implementation 'com.github.luben:zstd-jni:1.5.5-11'
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-csv:2.2.3'
+}
+
+protobuf {
+    protoc { artifact = "com.google.protobuf:protoc:${protocVersion}" }
+    plugins {
+        grpc { artifact = "io.grpc:protoc-gen-grpc-java:${grpcVersion}" }
+    }
+    generateProtoTasks {
+        all()*.plugins { grpc {} }
+    }
+}
+
+// Inform IDEs like IntelliJ IDEA, Eclipse or NetBeans about the generated code.
+sourceSets {
+    main {
+        java {
+            srcDirs 'build/generated/source/proto/main/grpc'
+            srcDirs 'build/generated/source/proto/main/java'
+        }
+    }
+}
+
+application {
+    mainClass = 'destination.JavaDestination'
+}
+
+tasks.register('copyProtos', Copy) {
+    from file("$rootDir/../..")
+    into file("src/main/proto/")
+    include "*.proto"
+}
+
+jar {
+    duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+
+    manifest {
+        attributes(
+            'Main-Class' : 'destination.JavaDestination'
+        )
+    }
+
+    from {
+        configurations.runtimeClasspath.filter{ it.exists() }.collect { it.isDirectory() ? it : zipTree(it) }
+    }
+}
diff --git a/v2_examples/destination/java/gradle/wrapper/gradle-wrapper.jar b/v2_examples/destination/java/gradle/wrapper/gradle-wrapper.jar
new file mode 100644
index 0000000..41d9927
Binary files /dev/null and b/v2_examples/destination/java/gradle/wrapper/gradle-wrapper.jar differ
diff --git a/v2_examples/destination/java/gradle/wrapper/gradle-wrapper.properties b/v2_examples/destination/java/gradle/wrapper/gradle-wrapper.properties
new file mode 100644
index 0000000..00e33ed
--- /dev/null
+++ b/v2_examples/destination/java/gradle/wrapper/gradle-wrapper.properties
@@ -0,0 +1,5 @@
+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.1-bin.zip
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists
diff --git a/v2_examples/destination/java/gradlew b/v2_examples/destination/java/gradlew
new file mode 100755
index 0000000..1b6c787
--- /dev/null
+++ b/v2_examples/destination/java/gradlew
@@ -0,0 +1,234 @@
+#!/bin/sh
+
+#
+# Copyright © 2015-2021 the original 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
+#
+#      https://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.
+#
+
+##############################################################################
+#
+#   Gradle start up script for POSIX generated by Gradle.
+#
+#   Important for running:
+#
+#   (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is
+#       noncompliant, but you have some other compliant shell such as ksh or
+#       bash, then to run this script, type that shell name before the whole
+#       command line, like:
+#
+#           ksh Gradle
+#
+#       Busybox and similar reduced shells will NOT work, because this script
+#       requires all of these POSIX shell features:
+#         * functions;
+#         * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
+#           «${var#prefix}», «${var%suffix}», and «$( cmd )»;
+#         * compound commands having a testable exit status, especially «case»;
+#         * various built-in commands including «command», «set», and «ulimit».
+#
+#   Important for patching:
+#
+#   (2) This script targets any POSIX shell, so it avoids extensions provided
+#       by Bash, Ksh, etc; in particular arrays are avoided.
+#
+#       The "traditional" practice of packing multiple parameters into a
+#       space-separated string is a well documented source of bugs and security
+#       problems, so this is (mostly) avoided, by progressively accumulating
+#       options in "$@", and eventually passing that to Java.
+#
+#       Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS,
+#       and GRADLE_OPTS) rely on word-splitting, this is performed explicitly;
+#       see the in-line comments for details.
+#
+#       There are tweaks for specific operating systems such as AIX, CygWin,
+#       Darwin, MinGW, and NonStop.
+#
+#   (3) This script is generated from the Groovy template
+#       https://github.com/gradle/gradle/blob/master/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+#       within the Gradle project.
+#
+#       You can find Gradle at https://github.com/gradle/gradle/.
+#
+##############################################################################
+
+# Attempt to set APP_HOME
+
+# Resolve links: $0 may be a link
+app_path=$0
+
+# Need this for daisy-chained symlinks.
+while
+    APP_HOME=${app_path%"${app_path##*/}"}  # leaves a trailing /; empty if no leading path
+    [ -h "$app_path" ]
+do
+    ls=$( ls -ld "$app_path" )
+    link=${ls#*' -> '}
+    case $link in             #(
+      /*)   app_path=$link ;; #(
+      *)    app_path=$APP_HOME$link ;;
+    esac
+done
+
+APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
+
+APP_NAME="Gradle"
+APP_BASE_NAME=${0##*/}
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD=maximum
+
+warn () {
+    echo "$*"
+} >&2
+
+die () {
+    echo
+    echo "$*"
+    echo
+    exit 1
+} >&2
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+nonstop=false
+case "$( uname )" in                #(
+  CYGWIN* )         cygwin=true  ;; #(
+  Darwin* )         darwin=true  ;; #(
+  MSYS* | MINGW* )  msys=true    ;; #(
+  NONSTOP* )        nonstop=true ;;
+esac
+
+CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+        # IBM's JDK on AIX uses strange locations for the executables
+        JAVACMD=$JAVA_HOME/jre/sh/java
+    else
+        JAVACMD=$JAVA_HOME/bin/java
+    fi
+    if [ ! -x "$JAVACMD" ] ; then
+        die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+    fi
+else
+    JAVACMD=java
+    which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+fi
+
+# Increase the maximum file descriptors if we can.
+if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
+    case $MAX_FD in #(
+      max*)
+        MAX_FD=$( ulimit -H -n ) ||
+            warn "Could not query maximum file descriptor limit"
+    esac
+    case $MAX_FD in  #(
+      '' | soft) :;; #(
+      *)
+        ulimit -n "$MAX_FD" ||
+            warn "Could not set maximum file descriptor limit to $MAX_FD"
+    esac
+fi
+
+# Collect all arguments for the java command, stacking in reverse order:
+#   * args from the command line
+#   * the main class name
+#   * -classpath
+#   * -D...appname settings
+#   * --module-path (only if needed)
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables.
+
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if "$cygwin" || "$msys" ; then
+    APP_HOME=$( cygpath --path --mixed "$APP_HOME" )
+    CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" )
+
+    JAVACMD=$( cygpath --unix "$JAVACMD" )
+
+    # Now convert the arguments - kludge to limit ourselves to /bin/sh
+    for arg do
+        if
+            case $arg in                                #(
+              -*)   false ;;                            # don't mess with options #(
+              /?*)  t=${arg#/} t=/${t%%/*}              # looks like a POSIX filepath
+                    [ -e "$t" ] ;;                      #(
+              *)    false ;;
+            esac
+        then
+            arg=$( cygpath --path --ignore --mixed "$arg" )
+        fi
+        # Roll the args list around exactly as many times as the number of
+        # args, so each arg winds up back in the position where it started, but
+        # possibly modified.
+        #
+        # NB: a `for` loop captures its iteration list before it begins, so
+        # changing the positional parameters here affects neither the number of
+        # iterations, nor the values presented in `arg`.
+        shift                   # remove old arg
+        set -- "$@" "$arg"      # push replacement arg
+    done
+fi
+
+# Collect all arguments for the java command;
+#   * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of
+#     shell script including quotes and variable substitutions, so put them in
+#     double quotes to make sure that they get re-expanded; and
+#   * put everything else in single quotes, so that it's not re-expanded.
+
+set -- \
+        "-Dorg.gradle.appname=$APP_BASE_NAME" \
+        -classpath "$CLASSPATH" \
+        org.gradle.wrapper.GradleWrapperMain \
+        "$@"
+
+# Use "xargs" to parse quoted args.
+#
+# With -n1 it outputs one arg per line, with the quotes and backslashes removed.
+#
+# In Bash we could simply go:
+#
+#   readarray ARGS < <( xargs -n1 <<<"$var" ) &&
+#   set -- "${ARGS[@]}" "$@"
+#
+# but POSIX shell has neither arrays nor command substitution, so instead we
+# post-process each arg (as a line of input to sed) to backslash-escape any
+# character that might be a shell metacharacter, then use eval to reverse
+# that process (while maintaining the separation between arguments), and wrap
+# the whole thing up as a single "set" statement.
+#
+# This will of course break if any of these variables contains a newline or
+# an unmatched quote.
+#
+
+eval "set -- $(
+        printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" |
+        xargs -n1 |
+        sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' |
+        tr '\n' ' '
+    )" '"$@"'
+
+exec "$JAVACMD" "$@"
diff --git a/v2_examples/destination/java/gradlew.bat b/v2_examples/destination/java/gradlew.bat
new file mode 100644
index 0000000..107acd3
--- /dev/null
+++ b/v2_examples/destination/java/gradlew.bat
@@ -0,0 +1,89 @@
+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem      https://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+
+@if "%DEBUG%" == "" @echo off
+@rem ##########################################################################
+@rem
+@rem  Gradle startup script for Windows
+@rem
+@rem ##########################################################################
+
+@rem Set local scope for the variables with windows NT shell
+if "%OS%"=="Windows_NT" setlocal
+
+set DIRNAME=%~dp0
+if "%DIRNAME%" == "" set DIRNAME=.
+set APP_BASE_NAME=%~n0
+set APP_HOME=%DIRNAME%
+
+@rem Resolve any "." and ".." in APP_HOME to make it shorter.
+for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi
+
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
+
+@rem Find java.exe
+if defined JAVA_HOME goto findJavaFromJavaHome
+
+set JAVA_EXE=java.exe
+%JAVA_EXE% -version >NUL 2>&1
+if "%ERRORLEVEL%" == "0" goto execute
+
+echo.
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:findJavaFromJavaHome
+set JAVA_HOME=%JAVA_HOME:"=%
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe
+
+if exist "%JAVA_EXE%" goto execute
+
+echo.
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:execute
+@rem Setup the command line
+
+set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
+
+
+@rem Execute Gradle
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %*
+
+:end
+@rem End local scope for the variables with windows NT shell
+if "%ERRORLEVEL%"=="0" goto mainEnd
+
+:fail
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
+rem the _cmd.exe /c_ return code!
+if  not "" == "%GRADLE_EXIT_CONSOLE%" exit 1
+exit /b 1
+
+:mainEnd
+if "%OS%"=="Windows_NT" endlocal
+
+:omega
diff --git a/v2_examples/destination/java/settings.gradle b/v2_examples/destination/java/settings.gradle
new file mode 100644
index 0000000..e1c0175
--- /dev/null
+++ b/v2_examples/destination/java/settings.gradle
@@ -0,0 +1,10 @@
+/*
+ * This file was generated by the Gradle 'init' task.
+ *
+ * The settings file is used to specify which projects to include in your build.
+ *
+ * Detailed information about configuring a multi-project build in Gradle can be found
+ * in the user manual at https://docs.gradle.org/7.4.1/userguide/multi_project_builds.html
+ */
+
+rootProject.name = 'JavaDestination'
diff --git a/v2_examples/destination/java/src/main/java/destination/DestinationServiceImpl.java b/v2_examples/destination/java/src/main/java/destination/DestinationServiceImpl.java
new file mode 100644
index 0000000..97eed80
--- /dev/null
+++ b/v2_examples/destination/java/src/main/java/destination/DestinationServiceImpl.java
@@ -0,0 +1,248 @@
+package destination;
+
+import fivetran_sdk.v2.*;
+import io.grpc.stub.StreamObserver;
+
+import java.util.Arrays;
+import java.util.Map;
+
+public class DestinationServiceImpl extends DestinationConnectorGrpc.DestinationConnectorImplBase {
+
+    private final String INFO = "INFO";
+    private final String WARNING = "WARNING";
+    private final String SEVERE = "SEVERE";
+
+    @Override
+    public void configurationForm(ConfigurationFormRequest request, StreamObserver<ConfigurationFormResponse> responseObserver) {
+        logMessage(INFO, "Fetching configuration form");
+        responseObserver.onNext(getConfigurationForm());
+
+        responseObserver.onCompleted();
+    }
+
+    private ConfigurationFormResponse getConfigurationForm() {
+
+        FormField writerType = FormField.newBuilder()
+                .setName("writerType")
+                .setLabel("Writer Type")
+                .setDescription("Choose the destination type")
+                .setDropdownField(
+                        DropdownField.newBuilder()
+                                .addAllDropdownField(Arrays.asList("Database", "File", "Cloud"))
+                                .build())
+                .setDefaultValue("Database")
+                .build();
+
+        FormField host = FormField.newBuilder()
+                .setName("host")
+                .setLabel("Host")
+                .setTextField(TextField.PlainText)
+                .setPlaceholder("your_host_details")
+                .build();
+
+        FormField port = FormField.newBuilder()
+                .setName("port")
+                .setLabel("Port")
+                .setTextField(TextField.PlainText)
+                .setPlaceholder("your_port_details")
+                .build();
+
+        FormField user = FormField.newBuilder()
+                .setName("user")
+                .setLabel("User")
+                .setTextField(TextField.PlainText)
+                .setPlaceholder("user_name")
+                .build();
+
+        FormField password = FormField.newBuilder()
+                .setName("password")
+                .setLabel("password")
+                .setTextField(TextField.Password)
+                .setPlaceholder("your_password")
+                .build();
+
+        FormField database = FormField.newBuilder()
+                .setName("database")
+                .setLabel("Database")
+                .setTextField(TextField.PlainText)
+                .setPlaceholder("your_database_name")
+                .build();
+
+        FormField table = FormField.newBuilder()
+                .setName("table")
+                .setLabel("Table")
+                .setTextField(TextField.PlainText)
+                .setPlaceholder("your_table_name")
+                .build();
+
+        FormField filePath = FormField.newBuilder()
+                .setName("filePath")
+                .setLabel("File Path")
+                .setTextField(TextField.PlainText)
+                .setPlaceholder("your_file_path")
+                .build();
+
+        FormField region = FormField.newBuilder()
+                .setName("region")
+                .setLabel("Cloud Region")
+                .setDescription("Choose the cloud region")
+                .setDropdownField(
+                        DropdownField.newBuilder()
+                                .addAllDropdownField(Arrays.asList("Azure", "AWS", "Google Cloud"))
+                                .build())
+                .setDefaultValue("Azure")
+                .build();
+
+        FormField enableEncryption = FormField.newBuilder()
+                .setName("enableEncryption")
+                .setDescription("To enable/disable encryption for data transfer")
+                .setLabel("Enable Encryption?")
+                .setToggleField(ToggleField.newBuilder().build())
+                .build();
+
+        // List of Visibility Conditions
+        VisibilityCondition visibilityConditionForCloud = VisibilityCondition.newBuilder()
+                .setConditionField("writerType")
+                .setStringValue("Cloud")
+                .build();
+
+        VisibilityCondition visibilityConditionForDatabase = VisibilityCondition.newBuilder()
+                .setConditionField("writerType")
+                .setStringValue("Database")
+                .build();
+
+        VisibilityCondition visibilityConditionForFile = VisibilityCondition.newBuilder()
+                .setConditionField("writerType")
+                .setStringValue("File")
+                .build();
+
+        // List of conditional fields
+        // Note: The 'name' and 'label' parameters in the FormField for conditional fields are not used.
+        FormField conditionalFieldForCloud = FormField.newBuilder()
+                .setName("conditionalFieldForCloud")
+                .setLabel("Conditional Field for Cloud")
+                .setConditionalFields(
+                        ConditionalFields.newBuilder()
+                                .setCondition(visibilityConditionForCloud)
+                                .addAllFields(Arrays.asList(host, port, user, password, region))
+                                .build())
+                .build();
+
+        FormField conditionalFieldForFile = FormField.newBuilder()
+                .setName("conditionalFieldForFile")
+                .setLabel("Conditional Field for File")
+                .setConditionalFields(
+                        ConditionalFields.newBuilder()
+                                .setCondition(visibilityConditionForFile)
+                                .addAllFields(Arrays.asList(host, port, user, password, table, filePath))
+                                .build())
+                .build();
+
+        FormField conditionalFieldForDatabase = FormField.newBuilder()
+                .setName("conditionalFieldForDatabase")
+                .setLabel("Conditional Field for Database")
+                .setConditionalFields(
+                        ConditionalFields.newBuilder()
+                                .setCondition(visibilityConditionForDatabase)
+                                .addAllFields(Arrays.asList(host, port, user, password, database,  table))
+                                .build())
+                .build();
+
+        return ConfigurationFormResponse.newBuilder()
+                .setSchemaSelectionSupported(true)
+                .setTableSelectionSupported(true)
+                .addAllFields(
+                        Arrays.asList(
+                                writerType,
+                                conditionalFieldForFile,
+                                conditionalFieldForCloud,
+                                conditionalFieldForDatabase,
+                                enableEncryption))
+                .addAllTests(
+                        Arrays.asList(
+                                ConfigurationTest.newBuilder().setName("connect").setLabel("Tests connection").build(),
+                                ConfigurationTest.newBuilder().setName("select").setLabel("Tests selection").build()))
+                .build();
+    }
+
+    @Override
+    public void test(TestRequest request, StreamObserver<TestResponse> responseObserver) {
+        Map<String, String> configuration = request.getConfigurationMap();
+        String testName = request.getName();
+        String message = String.format("Test Name: %s", testName);
+        logMessage(INFO, message);
+
+        responseObserver.onNext(TestResponse.newBuilder().setSuccess(true).build());
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void describeTable(DescribeTableRequest request, StreamObserver<DescribeTableResponse> responseObserver) {
+        Map<String, String> configuration = request.getConfigurationMap();
+
+        DescribeTableResponse response = DescribeTableResponse.newBuilder()
+                .setTable(
+                        Table.newBuilder()
+                                .setName(request.getTableName())
+                                .addAllColumns(
+                                Arrays.asList(
+                                        Column.newBuilder().setName("a1").setType(DataType.UNSPECIFIED).setPrimaryKey(true).build(),
+                                        Column.newBuilder().setName("a2").setType(DataType.DOUBLE).build())
+                        ).build()).build();
+
+        responseObserver.onNext(response);
+        logMessage(SEVERE, "Sample Severe log: Completed describe Table method");
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void createTable(CreateTableRequest request, StreamObserver<CreateTableResponse> responseObserver) {
+        Map<String, String> configuration = request.getConfigurationMap();
+
+        String message = "[CreateTable]: "
+                + request.getSchemaName() + " | " + request.getTable().getName() + " | " + request.getTable().getColumnsList();
+        logMessage(INFO, message);
+        responseObserver.onNext(CreateTableResponse.newBuilder().setSuccess(true).build());
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void alterTable(AlterTableRequest request, StreamObserver<AlterTableResponse> responseObserver) {
+        Map<String, String> configuration = request.getConfigurationMap();
+
+        String message = "[AlterTable]: " +
+                request.getSchemaName() + " | " + request.getTable().getName() + " | " + request.getTable().getColumnsList();
+        logMessage(INFO, message);
+        responseObserver.onNext(AlterTableResponse.newBuilder().setSuccess(true).build());
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void truncate(TruncateRequest request, StreamObserver<TruncateResponse> responseObserver) {
+        System.out.printf("[TruncateTable]: %s | %s | soft=%s%n",
+                request.getSchemaName(), request.getTableName(), request.hasSoft());
+        responseObserver.onNext(TruncateResponse.newBuilder().setSuccess(true).build());
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void writeBatch(WriteBatchRequest request, StreamObserver<WriteBatchResponse> responseObserver) {
+        String message = "[WriteBatch]: " + request.getSchemaName() + " | " + request.getTable().getName();
+        logMessage(WARNING, String.format("Sample severe message: %s", message));
+        for (String file : request.getReplaceFilesList()) {
+            System.out.println("Replace files: " + file);
+        }
+        for (String file : request.getUpdateFilesList()) {
+            System.out.println("Update files: " + file);
+        }
+        for (String file : request.getDeleteFilesList()) {
+            System.out.println("Delete files: " + file);
+        }
+        responseObserver.onNext(WriteBatchResponse.newBuilder().setSuccess(true).build());
+        responseObserver.onCompleted();
+    }
+
+    private void logMessage(String level, String message){
+        System.out.println(String.format("{\"level\":\"%s\", \"message\": \"%s\", \"message-origin\": \"sdk_destination\"}", level, message));
+    }
+}
diff --git a/v2_examples/destination/java/src/main/java/destination/JavaDestination.java b/v2_examples/destination/java/src/main/java/destination/JavaDestination.java
new file mode 100644
index 0000000..95ad99c
--- /dev/null
+++ b/v2_examples/destination/java/src/main/java/destination/JavaDestination.java
@@ -0,0 +1,22 @@
+package destination;
+
+import io.grpc.*;
+
+import java.io.IOException;
+
+/**
+ * Example Plugin Connector (gRPC server)
+ * In production, it will be stored as a container image
+ */
+public class JavaDestination {
+
+    public static void main(String[] args) throws InterruptedException, IOException {
+        Server server = ServerBuilder
+                .forPort(50052)
+                .addService(new DestinationServiceImpl()).build();
+
+        server.start();
+        System.out.println("Destination gRPC server started");
+        server.awaitTermination();
+    }
+}
diff --git a/v2_examples/destination/python/README.md b/v2_examples/destination/python/README.md
new file mode 100644
index 0000000..f154a71
--- /dev/null
+++ b/v2_examples/destination/python/README.md
@@ -0,0 +1,15 @@
+# Python Destination Example
+
+## Pre-requisites
+- Python 3.9 or later
+
+## Steps
+- Run the build.sh file to copy protos, install python dependencies in virtual environment
+```commandline
+sh build.sh
+```
+
+- Execute `run.sh` to run the connector
+```commandline
+sh run.sh
+```
\ No newline at end of file
diff --git a/v2_examples/destination/python/build.sh b/v2_examples/destination/python/build.sh
new file mode 100755
index 0000000..d318efd
--- /dev/null
+++ b/v2_examples/destination/python/build.sh
@@ -0,0 +1,29 @@
+#!/bin/bash
+
+#Create virtual environment
+python3 -m venv destination_run
+
+#Activate virtual environment
+source destination_run/bin/activate
+
+# Make a directory protos
+mkdir -p protos
+
+# Copy proto files t oprotos directory
+cp ../../*.proto protos/
+
+# Install the required packages
+pip install -r requirements.txt
+
+# Make a directory sdk_pb2
+mkdir -p sdk_pb2
+
+# Generate grpc python code and store it in sdk_pb2
+python -m grpc_tools.protoc \
+       --proto_path=./protos/ \
+       --python_out=sdk_pb2 \
+       --pyi_out=sdk_pb2 \
+       --grpc_python_out=sdk_pb2 protos/*.proto
+
+# Deactivate virtual environment
+deactivate
\ No newline at end of file
diff --git a/v2_examples/destination/python/main.py b/v2_examples/destination/python/main.py
new file mode 100644
index 0000000..8a0f43e
--- /dev/null
+++ b/v2_examples/destination/python/main.py
@@ -0,0 +1,241 @@
+from concurrent import futures
+import grpc
+import read_csv
+import sys
+sys.path.append('sdk_pb2')
+
+from sdk_pb2 import destination_sdk_v2_pb2
+from sdk_pb2 import common_v2_pb2
+from sdk_pb2 import destination_sdk_v2_pb2_grpc
+
+
+INFO = "INFO"
+WARNING = "WARNING"
+SEVERE = "SEVERE"
+
+class DestinationImpl(destination_sdk_v2_pb2_grpc.DestinationConnectorServicer):
+    def ConfigurationForm(self, request, context):
+        log_message(INFO, "Fetching Configuration form")
+
+        # Create the form fields
+        form_fields = common_v2_pb2.ConfigurationFormResponse(
+            schema_selection_supported=True,
+            table_selection_supported=True
+        )
+
+        # writerType field with dropdown
+        writer_type = common_v2_pb2.FormField(
+            name="writerType",
+            label="Writer Type",
+            description="Choose the destination type",
+            dropdown_field=common_v2_pb2.DropdownField(dropdown_field=["Database", "File", "Cloud"]),
+            default_value="Database"
+        )
+
+        # host field
+        host = common_v2_pb2.FormField(
+            name="host",
+            label="Host",
+            text_field=common_v2_pb2.TextField.PlainText,
+            placeholder="your_host_details"
+        )
+
+        # port field
+        port = common_v2_pb2.FormField(
+            name="port",
+            label="Port",
+            text_field=common_v2_pb2.TextField.PlainText,
+            placeholder="your_port_details"
+        )
+
+        # user field
+        user = common_v2_pb2.FormField(
+            name="user",
+            label="User",
+            text_field=common_v2_pb2.TextField.PlainText,
+            placeholder="user_name"
+        )
+
+        # password field
+        password = common_v2_pb2.FormField(
+            name="password",
+            label="Password",
+            text_field=common_v2_pb2.TextField.Password,
+            placeholder="your_password"
+        )
+
+        # database field
+        database = common_v2_pb2.FormField(
+            name="database",
+            label="Database",
+            text_field=common_v2_pb2.TextField.PlainText,
+            placeholder="your_database_name"
+        )
+
+        # table field
+        table = common_v2_pb2.FormField(
+            name="table",
+            label="Table",
+            text_field=common_v2_pb2.TextField.PlainText,
+            placeholder="your_table_name"
+        )
+
+        # filePath field
+        file_path = common_v2_pb2.FormField(
+            name="filePath",
+            label="File Path",
+            text_field=common_v2_pb2.TextField.PlainText,
+            placeholder="your_file_path"
+        )
+
+        # region field with dropdown
+        region = common_v2_pb2.FormField(
+            name="region",
+            label="Cloud Region",
+            description="Choose the cloud region",
+            dropdown_field=common_v2_pb2.DropdownField(dropdown_field=["Azure", "AWS", "Google Cloud"]),
+            default_value="Azure"
+        )
+
+        # enableEncryption toggle field
+        enable_encryption = common_v2_pb2.FormField(
+            name="enableEncryption",
+            label="Enable Encryption?",
+            description="To enable/disable encryption for data transfer",
+            toggle_field=common_v2_pb2.ToggleField()
+        )
+
+        # Define Visibility Conditions for Conditional Fields
+        visibility_condition_for_cloud = common_v2_pb2.VisibilityCondition(
+            condition_field="writerType",
+            string_value="Cloud"
+        )
+
+        visibility_condition_for_database = common_v2_pb2.VisibilityCondition(
+            condition_field="writerType",
+            string_value="Database"
+        )
+
+        visibility_condition_for_file = common_v2_pb2.VisibilityCondition(
+            condition_field="writerType",
+            string_value="File"
+        )
+
+        # List of conditional fields
+        # Note: The 'name' and 'label' parameters in the FormField for conditional fields are not used.
+
+        # Create conditional fields for Cloud
+        conditional_fields_for_cloud = common_v2_pb2.ConditionalFields(
+            condition=visibility_condition_for_cloud,
+            fields=[host, port, user, password, region]
+        )
+
+        # Create conditional fields for File
+        conditional_fields_for_file = common_v2_pb2.ConditionalFields(
+            condition=visibility_condition_for_file,
+            fields=[host, port, user, password, table, file_path]
+        )
+
+        # Create conditional fields for Database
+        conditional_fields_for_database = common_v2_pb2.ConditionalFields(
+            condition=visibility_condition_for_database,
+            fields=[host, port, user, password, database, table]
+        )
+
+        # Add conditional fields to the form
+        conditional_field_for_cloud = common_v2_pb2.FormField(
+            name="conditional_field_for_cloud",
+            label="Conditional field for cloud",
+            conditional_fields=conditional_fields_for_cloud
+        )
+
+        conditional_field_for_file = common_v2_pb2.FormField(
+            name="conditional_field_for_file",
+            label="Conditional field for File",
+            conditional_fields=conditional_fields_for_file
+        )
+
+        conditional_field_for_database = common_v2_pb2.FormField(
+            name="conditional_field_for_database",
+            label="Conditional field for Database",
+            conditional_fields=conditional_fields_for_database
+        )
+
+        # Add all fields to the form response
+        form_fields.fields.extend([
+            writer_type,
+            conditional_field_for_file,
+            conditional_field_for_cloud,
+            conditional_field_for_database,
+            enable_encryption
+        ])
+
+        # Add tests to the form
+        form_fields.tests.add(
+            name="connect",
+            label="Tests connection"
+        )
+
+        form_fields.tests.add(
+            name="select",
+            label="Tests selection"
+        )
+
+        return form_fields
+
+    def Test(self, request, context):
+        test_name = request.name
+        log_message(INFO, "test name: " + test_name)
+        return common_v2_pb2.TestResponse(success=True)
+
+    def CreateTable(self, request, context):
+        print("[CreateTable] :" + str(request.schema_name) + " | " + str(request.table.name) + " | " + str(request.table.columns))
+        return destination_sdk_v2_pb2.CreateTableResponse(success=True)
+
+    def AlterTable(self, request, context):
+        res: destination_sdk_v2_pb2.AlterTableResponse
+
+        print("[AlterTable]: " + str(request.schema_name) + " | " + str(request.table.name) + " | " + str(request.table.columns))
+        return destination_sdk_v2_pb2.AlterTableResponse(success=True)
+
+    def Truncate(self, request, context):
+        print("[TruncateTable]: " + str(request.schema_name) + " | " + str(request.schema_name) + " | soft" + str(request.soft))
+        return destination_sdk_v2_pb2.TruncateResponse(success=True)
+
+    def WriteBatch(self, request, context):
+        for replace_file in request.replace_files:
+            print("replace files: " + str(replace_file))
+        for update_file in request.update_files:
+            print("replace files: " + str(update_file))
+        for delete_file in request.delete_files:
+            print("replace files: " + str(delete_file))
+
+        log_message(WARNING, "Data loading started for table " + request.table.name)
+        for key, value in request.keys.items():
+            print("----------------------------------------------------------------------------")
+            print("Decrypting and printing file :" + str(key))
+            print("----------------------------------------------------------------------------")
+            read_csv.decrypt_file(key, value)
+        log_message(INFO, "\nData loading completed for table " + request.table.name + "\n")
+
+        res: destination_sdk_v2_pb2.WriteBatchResponse = destination_sdk_v2_pb2.WriteBatchResponse(success=True)
+        return res
+
+    def DescribeTable(self, request, context):
+        column1 = common_v2_pb2.Column(name="a1", type=common_v2_pb2.DataType.UNSPECIFIED, primary_key=True)
+        column2 = common_v2_pb2.Column(name="a2", type=common_v2_pb2.DataType.DOUBLE)
+        table: common_v2_pb2.Table = common_v2_pb2.Table(name=request.table_name, columns=[column1, column2])
+        log_message(SEVERE, "Sample severe message: Completed fetching table info")
+        return destination_sdk_v2_pb2.DescribeTableResponse(not_found=False, table=table)
+
+def log_message(level, message):
+    print(f'{{"level":"{level}", "message": "{message}", "message-origin": "sdk_destination"}}')
+
+if __name__ == '__main__':
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
+    server.add_insecure_port('[::]:50052')
+    destination_sdk_v2_pb2_grpc.add_DestinationConnectorServicer_to_server(DestinationImpl(), server)
+    server.start()
+    print("Destination gRPC server started...")
+    server.wait_for_termination()
+    print("Destination gRPC server terminated...")
diff --git a/v2_examples/destination/python/read_csv.py b/v2_examples/destination/python/read_csv.py
new file mode 100644
index 0000000..56a9216
--- /dev/null
+++ b/v2_examples/destination/python/read_csv.py
@@ -0,0 +1,32 @@
+from zstandard import ZstdDecompressor
+from Crypto.Cipher import AES
+import csv
+
+
+# AES decryption function
+def aes_decrypt(key, ciphertext):
+    cipher = AES.new(key, AES.MODE_CBC, iv=ciphertext[:AES.block_size])
+    plaintext = cipher.decrypt(ciphertext[AES.block_size:])
+    return plaintext.rstrip(b'\0')
+
+
+# Zstandard decompression function
+def zstd_decompress(compressed_data):
+    decompressor = ZstdDecompressor()
+    decompressed_data = decompressor.decompressobj().decompress(compressed_data)
+    return decompressed_data
+
+
+# Read the encrypted and compressed data
+def decrypt_file(input_file_path, value):
+    with open(input_file_path, 'rb') as file:
+        encrypted_and_compressed_data = file.read()
+        decrypted_data = aes_decrypt(value, encrypted_and_compressed_data)
+        decompressed_data = zstd_decompress(decrypted_data)
+        csv_data = decompressed_data.decode('utf-8')
+        csv_reader = csv.reader(csv_data.splitlines())
+        headers = next(csv_reader)
+        print(f"{'  |  '.join(headers)}")
+        print('-' * (len(headers) * 15))
+        for row in csv_reader:
+            print(f"{'  |  '.join(row)}")
diff --git a/v2_examples/destination/python/requirements.txt b/v2_examples/destination/python/requirements.txt
new file mode 100644
index 0000000..9e5b8e3
--- /dev/null
+++ b/v2_examples/destination/python/requirements.txt
@@ -0,0 +1,8 @@
+grpcio==1.60.1
+grpcio-tools==1.60.1
+protobuf==4.25.3
+google~=3.0.0
+pip~=23.0.1
+setuptools~=65.5.0
+zstandard~=0.22.0
+pycryptodome==3.19.1
\ No newline at end of file
diff --git a/v2_examples/destination/python/run.sh b/v2_examples/destination/python/run.sh
new file mode 100755
index 0000000..d90200f
--- /dev/null
+++ b/v2_examples/destination/python/run.sh
@@ -0,0 +1,3 @@
+source destination_run/bin/activate
+python main.py
+deactivate
\ No newline at end of file