geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kl...@apache.org
Subject [geode] 01/01: GEODE-3781: add geode-connectors module
Date Fri, 17 Nov 2017 20:53:06 GMT
This is an automated email from the ASF dual-hosted git repository.

klund pushed a commit to branch feature/GEODE-3781
in repository https://gitbox.apache.org/repos/asf/geode.git

commit 69337aaa6bca5274cced16a4a776b6f04709b761
Author: Anil <agingade@pivotal.io>
AuthorDate: Mon Oct 23 12:01:58 2017 -0700

    GEODE-3781: add geode-connectors module
    
    Introduce new PDX JDBC Connector Service
    * GEODE-3782: implement JdbcWriter
    * GEODE-3783: implement JdbcAsyncWriter
    * GEODE-3823: implement JdbcLoader
---
 geode-assembly/build.gradle                        |   6 +-
 geode-connectors/build.gradle                      |  32 ++
 .../geode/connectors/jdbc/JdbcAsyncWriter.java     | 107 ++++++
 .../apache/geode/connectors/jdbc/JdbcLoader.java   |  54 +++
 .../apache/geode/connectors/jdbc/JdbcWriter.java   | 102 ++++++
 .../jdbc/internal/AbstractJdbcCallback.java        |  63 ++++
 .../connectors/jdbc/internal/ColumnValue.java      |  44 +++
 .../jdbc/internal/ConnectionConfiguration.java     |  84 +++++
 .../jdbc/internal/ConnectionManager.java           | 193 +++++++++++
 .../internal/InternalJdbcConnectorService.java     |  30 ++
 .../jdbc/internal/JdbcConnectorService.java        |  92 ++++++
 .../jdbc/internal/PreparedStatementCache.java      | 104 ++++++
 .../connectors/jdbc/internal/RegionMapping.java    | 118 +++++++
 .../geode/connectors/jdbc/internal/SqlHandler.java | 187 +++++++++++
 .../jdbc/internal/SqlStatementFactory.java         |  78 +++++
 .../jdbc/internal/xml/ConnectionConfigBuilder.java |  48 +++
 .../connectors/jdbc/internal/xml/ElementType.java  | 131 ++++++++
 .../xml/JdbcConnectorServiceXmlGenerator.java      | 130 ++++++++
 .../xml/JdbcConnectorServiceXmlParser.java         |  57 ++++
 .../internal/xml/JdbcServiceConfiguration.java     |  60 ++++
 .../jdbc/internal/xml/RegionMappingBuilder.java    |  65 ++++
 .../org.apache.geode.internal.cache.CacheService   |   1 +
 ....apache.geode.internal.cache.xmlcache.XmlParser |   1 +
 .../org/apache/geode/connectors/jdbc/Employee.java |  54 +++
 .../jdbc/JdbcAsyncWriterIntegrationTest.java       | 254 ++++++++++++++
 .../geode/connectors/jdbc/JdbcAsyncWriterTest.java |  93 ++++++
 .../connectors/jdbc/JdbcLoaderIntegrationTest.java | 105 ++++++
 .../geode/connectors/jdbc/JdbcLoaderTest.java      |  44 +++
 .../connectors/jdbc/JdbcWriterIntegrationTest.java | 234 +++++++++++++
 .../geode/connectors/jdbc/JdbcWriterTest.java      | 119 +++++++
 .../jdbc/internal/AbstractJdbcCallbackTest.java    |  73 +++++
 .../connectors/jdbc/internal/ColumnValueTest.java  |  57 ++++
 .../jdbc/internal/ConnectionConfigurationTest.java |  63 ++++
 .../jdbc/internal/ConnectionManagerUnitTest.java   | 252 ++++++++++++++
 .../jdbc/internal/JdbcConnectorServiceTest.java    |  88 +++++
 .../jdbc/internal/PreparedStatementCacheTest.java  |  82 +++++
 .../jdbc/internal/RegionMappingTest.java           | 108 ++++++
 .../connectors/jdbc/internal/SqlHandlerTest.java   | 365 +++++++++++++++++++++
 .../jdbc/internal/SqlStatementFactoryTest.java     |  81 +++++
 .../jdbc/internal/TestConfigService.java           |  54 +++
 .../jdbc/internal/TestableConnectionManager.java   |  22 ++
 .../internal/xml/ConnectionConfigBuilderTest.java  |  46 +++
 .../jdbc/internal/xml/ElementTypeTest.java         | 216 ++++++++++++
 ...onnectorServiceXmlGeneratorIntegrationTest.java | 195 +++++++++++
 .../xml/JdbcConnectorServiceXmlGeneratorTest.java  |  33 ++
 .../JdbcConnectorServiceXmlIntegrationTest.java    | 116 +++++++
 .../xml/JdbcConnectorServiceXmlParserTest.java     | 106 ++++++
 .../internal/xml/JdbcServiceConfigurationTest.java | 136 ++++++++
 .../internal/xml/RegionMappingBuilderTest.java     |  54 +++
 .../java/org/apache/geode/test/fake/Fakes.java     |   5 +-
 settings.gradle                                    |   1 +
 51 files changed, 4841 insertions(+), 2 deletions(-)

diff --git a/geode-assembly/build.gradle b/geode-assembly/build.gradle
index 4615a6b..542ab45 100755
--- a/geode-assembly/build.gradle
+++ b/geode-assembly/build.gradle
@@ -59,6 +59,7 @@ dependencies {
   archives project(':geode-client-protocol')
   archives project(':geode-json')
   archives project(':geode-core')
+  archives project(':geode-connectors')
   archives project(':geode-lucene')
   archives project(':geode-old-client-support')
   archives project(':geode-protobuf')
@@ -362,7 +363,10 @@ distributions {
 
         from project(":geode-lucene").configurations.runtime
         from project(":geode-lucene").configurations.archives.allArtifacts.files
-        
+       
+        from project(":geode-connectors").configurations.runtime
+        from project(":geode-connectors").configurations.archives.allArtifacts.files
+ 
         from project(":geode-old-client-support").configurations.runtime
         from project(":geode-old-client-support").configurations.archives.allArtifacts.files
 
diff --git a/geode-connectors/build.gradle b/geode-connectors/build.gradle
new file mode 100644
index 0000000..b11f352
--- /dev/null
+++ b/geode-connectors/build.gradle
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+dependencies {
+    compile project(':geode-core')
+    compile project(':geode-common')
+
+    testCompile project(':geode-junit')
+
+    //Connectors test framework.
+    testRuntime 'org.apache.derby:derby:' + project.'derby.version'
+    testCompile 'com.pholser:junit-quickcheck-core:' + project.'junit-quickcheck.version'
+    testCompile 'com.pholser:junit-quickcheck-generators:' + project.'junit-quickcheck.version'
+    testCompile files(project(':geode-core').sourceSets.test.output)
+    testCompile project(':geode-old-versions')
+}
+
+integrationTest.forkEvery 0
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriter.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriter.java
new file mode 100644
index 0000000..465e1c6
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriter.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.CopyHelper;
+import org.apache.geode.annotations.Experimental;
+import org.apache.geode.cache.asyncqueue.AsyncEvent;
+import org.apache.geode.cache.asyncqueue.AsyncEventListener;
+import org.apache.geode.cache.query.internal.DefaultQuery;
+import org.apache.geode.connectors.jdbc.internal.AbstractJdbcCallback;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.pdx.PdxInstance;
+
+/**
+ * This class provides write behind cache semantics for a JDBC data source using AsyncEventListener.
+ *
+ * @since Geode 1.4
+ */
+@Experimental
+public class JdbcAsyncWriter extends AbstractJdbcCallback implements AsyncEventListener {
+  private static final Logger logger = LogService.getLogger();
+
+  private AtomicLong totalEvents = new AtomicLong();
+  private AtomicLong successfulEvents = new AtomicLong();
+
+  @SuppressWarnings("unused")
+  public JdbcAsyncWriter() {
+    super();
+  }
+
+  // Constructor for test purposes only
+  JdbcAsyncWriter(SqlHandler sqlHandler) {
+    super(sqlHandler);
+  }
+
+  @Override
+  public boolean processEvents(List<AsyncEvent> events) {
+    changeTotalEvents(events.size());
+
+    if (!events.isEmpty()) {
+      checkInitialized((InternalCache) events.get(0).getRegion().getRegionService());
+    }
+
+    DefaultQuery.setPdxReadSerialized(true);
+    try {
+      for (AsyncEvent event : events) {
+        try {
+          getSqlHandler().write(event.getRegion(), event.getOperation(), event.getKey(),
+              getPdxInstance(event));
+          changeSuccessfulEvents(1);
+        } catch (RuntimeException ex) {
+          logger.error("Exception processing event {}", event, ex);
+        }
+      }
+    } finally {
+      DefaultQuery.setPdxReadSerialized(false);
+    }
+
+    return true;
+  }
+
+  long getTotalEvents() {
+    return totalEvents.get();
+  }
+
+  long getSuccessfulEvents() {
+    return successfulEvents.get();
+  }
+
+  private void changeSuccessfulEvents(long delta) {
+    successfulEvents.addAndGet(delta);
+  }
+
+  private void changeTotalEvents(long delta) {
+    totalEvents.addAndGet(delta);
+  }
+
+  /**
+   * precondition: DefaultQuery.setPdxReadSerialized(true)
+   */
+  private PdxInstance getPdxInstance(AsyncEvent event) {
+    Object value = event.getDeserializedValue();
+    if (!(value instanceof PdxInstance)) {
+      value = CopyHelper.copy(value);
+    }
+    return (PdxInstance) value;
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcLoader.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcLoader.java
new file mode 100644
index 0000000..f3ba9a5
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcLoader.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import org.apache.geode.annotations.Experimental;
+import org.apache.geode.cache.CacheLoader;
+import org.apache.geode.cache.CacheLoaderException;
+import org.apache.geode.cache.LoaderHelper;
+import org.apache.geode.connectors.jdbc.internal.AbstractJdbcCallback;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.internal.cache.InternalCache;
+
+/**
+ * This class provides loading from a data source using JDBC.
+ *
+ * @since Geode 1.4
+ */
+@Experimental
+public class JdbcLoader<K, V> extends AbstractJdbcCallback implements CacheLoader<K, V> {
+
+  @SuppressWarnings("unused")
+  public JdbcLoader() {
+    super();
+  }
+
+  // Constructor for test purposes only
+  JdbcLoader(SqlHandler sqlHandler) {
+    super(sqlHandler);
+  }
+
+  /**
+   * @return this method always returns a PdxInstance. It does not matter what the V generic
+   *         parameter is set to.
+   */
+  @Override
+  public V load(LoaderHelper<K, V> helper) throws CacheLoaderException {
+    // The following cast to V is to keep the compiler happy
+    // but is erased at runtime and no actual cast happens.
+    checkInitialized((InternalCache) helper.getRegion().getRegionService());
+    return (V) getSqlHandler().read(helper.getRegion(), helper.getKey());
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcWriter.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcWriter.java
new file mode 100644
index 0000000..c901d3d
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JdbcWriter.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import org.apache.geode.CopyHelper;
+import org.apache.geode.annotations.Experimental;
+import org.apache.geode.cache.CacheWriter;
+import org.apache.geode.cache.CacheWriterException;
+import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.RegionEvent;
+import org.apache.geode.cache.SerializedCacheValue;
+import org.apache.geode.cache.query.internal.DefaultQuery;
+import org.apache.geode.connectors.jdbc.internal.AbstractJdbcCallback;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.pdx.PdxInstance;
+
+/**
+ * This class provides synchronous write through to a data source using JDBC.
+ *
+ * @since Geode 1.4
+ */
+@Experimental
+public class JdbcWriter<K, V> extends AbstractJdbcCallback implements CacheWriter<K, V> {
+
+  @SuppressWarnings("unused")
+  public JdbcWriter() {
+    super();
+  }
+
+  // Constructor for test purposes only
+  JdbcWriter(SqlHandler sqlHandler) {
+    super(sqlHandler);
+  }
+
+
+  @Override
+  public void beforeUpdate(EntryEvent<K, V> event) throws CacheWriterException {
+    checkInitialized((InternalCache) event.getRegion().getRegionService());
+    getSqlHandler().write(event.getRegion(), event.getOperation(), event.getKey(),
+        getPdxNewValue(event));
+  }
+
+  @Override
+  public void beforeCreate(EntryEvent<K, V> event) throws CacheWriterException {
+    checkInitialized((InternalCache) event.getRegion().getRegionService());
+    getSqlHandler().write(event.getRegion(), event.getOperation(), event.getKey(),
+        getPdxNewValue(event));
+  }
+
+  @Override
+  public void beforeDestroy(EntryEvent<K, V> event) throws CacheWriterException {
+    checkInitialized((InternalCache) event.getRegion().getRegionService());
+    getSqlHandler().write(event.getRegion(), event.getOperation(), event.getKey(),
+        getPdxNewValue(event));
+  }
+
+  @Override
+  public void beforeRegionDestroy(RegionEvent<K, V> event) throws CacheWriterException {
+    // this event is not sent to JDBC
+  }
+
+  @Override
+  public void beforeRegionClear(RegionEvent<K, V> event) throws CacheWriterException {
+    // this event is not sent to JDBC
+  }
+
+  private PdxInstance getPdxNewValue(EntryEvent<K, V> event) {
+    DefaultQuery.setPdxReadSerialized(true);
+    try {
+      Object newValue = event.getNewValue();
+      if (!(newValue instanceof PdxInstance)) {
+        SerializedCacheValue<V> serializedNewValue = event.getSerializedNewValue();
+        if (serializedNewValue != null) {
+          newValue = serializedNewValue.getDeserializedValue();
+        } else {
+          newValue = CopyHelper.copy(newValue);
+        }
+        if (newValue != null && !(newValue instanceof PdxInstance)) {
+          String valueClassName = newValue == null ? "null" : newValue.getClass().getName();
+          throw new IllegalArgumentException(getClass().getSimpleName()
+              + " only supports PDX values; newValue is " + valueClassName);
+        }
+      }
+      return (PdxInstance) newValue;
+    } finally {
+      DefaultQuery.setPdxReadSerialized(false);
+    }
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/AbstractJdbcCallback.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/AbstractJdbcCallback.java
new file mode 100644
index 0000000..478edd0
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/AbstractJdbcCallback.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import java.util.Properties;
+
+import org.apache.geode.cache.CacheCallback;
+import org.apache.geode.internal.cache.InternalCache;
+
+public abstract class AbstractJdbcCallback implements CacheCallback {
+
+  private volatile SqlHandler sqlHandler;
+
+  protected AbstractJdbcCallback() {
+    // nothing
+  }
+
+  protected AbstractJdbcCallback(SqlHandler sqlHandler) {
+    this.sqlHandler = sqlHandler;
+  }
+
+  @Override
+  public void close() {
+    if (sqlHandler != null) {
+      sqlHandler.close();
+    }
+  }
+
+  @Override
+  public void init(Properties props) {
+    // nothing
+  }
+
+  protected SqlHandler getSqlHandler() {
+    return sqlHandler;
+  }
+
+  protected void checkInitialized(InternalCache cache) {
+    if (sqlHandler == null) {
+      initialize(cache);
+    }
+  }
+
+  private synchronized void initialize(InternalCache cache) {
+    if (sqlHandler == null) {
+      InternalJdbcConnectorService service = cache.getService(InternalJdbcConnectorService.class);
+      ConnectionManager manager = new ConnectionManager(service);
+      sqlHandler = new SqlHandler(manager);
+    }
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ColumnValue.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ColumnValue.java
new file mode 100644
index 0000000..c3f44d0
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ColumnValue.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+class ColumnValue {
+  private final boolean isKey;
+  private final String columnName;
+  private final Object value;
+
+  ColumnValue(boolean isKey, String columnName, Object value) {
+    this.isKey = isKey;
+    this.columnName = columnName;
+    this.value = value;
+  }
+
+  boolean isKey() {
+    return isKey;
+  }
+
+  String getColumnName() {
+    return columnName;
+  }
+
+  Object getValue() {
+    return value;
+  }
+
+  @Override
+  public String toString() {
+    return "ColumnValue [isKey=" + isKey + ", columnName=" + columnName + ", value=" + value + "]";
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ConnectionConfiguration.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ConnectionConfiguration.java
new file mode 100644
index 0000000..fa3807d
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ConnectionConfiguration.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+public class ConnectionConfiguration {
+
+  private final String name;
+  private final String url;
+  private final String user;
+  private final String password;
+
+  public ConnectionConfiguration(String name, String url, String user, String password) {
+    this.name = name;
+    this.url = url;
+    this.user = user;
+    this.password = password;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public String getUrl() {
+    return url;
+  }
+
+  public String getUser() {
+    return user;
+  }
+
+  public String getPassword() {
+    return password;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    ConnectionConfiguration that = (ConnectionConfiguration) o;
+
+    if (name != null ? !name.equals(that.name) : that.name != null) {
+      return false;
+    }
+    if (url != null ? !url.equals(that.url) : that.url != null) {
+      return false;
+    }
+    if (user != null ? !user.equals(that.user) : that.user != null) {
+      return false;
+    }
+    return password != null ? password.equals(that.password) : that.password == null;
+  }
+
+  @Override
+  public int hashCode() {
+    int result = name != null ? name.hashCode() : 0;
+    result = 31 * result + (url != null ? url.hashCode() : 0);
+    result = 31 * result + (user != null ? user.hashCode() : 0);
+    result = 31 * result + (password != null ? password.hashCode() : 0);
+    return result;
+  }
+
+  @Override
+  public String toString() {
+    return "ConnectionConfiguration{" + "name='" + name + '\'' + ", url='" + url + '\'' + ", user='"
+        + user + '\'' + ", password='" + password + '\'' + '}';
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ConnectionManager.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ConnectionManager.java
new file mode 100644
index 0000000..d382112
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/ConnectionManager.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.geode.cache.Operation;
+import org.apache.geode.pdx.PdxInstance;
+
+class ConnectionManager {
+
+  private final InternalJdbcConnectorService configService;
+  private final Map<String, Connection> connectionMap = new ConcurrentHashMap<>();
+  private final ConcurrentMap<String, String> tableToPrimaryKeyMap = new ConcurrentHashMap<>();
+  private final ThreadLocal<PreparedStatementCache> preparedStatementCache = new ThreadLocal<>();
+
+  ConnectionManager(InternalJdbcConnectorService configService) {
+    this.configService = configService;
+  }
+
+  RegionMapping getMappingForRegion(String regionName) {
+    return configService.getMappingForRegion(regionName);
+  }
+
+  Connection getConnection(ConnectionConfiguration config) {
+    Connection connection = connectionMap.get(config.getName());
+    try {
+      if (connection != null && !connection.isClosed()) {
+        return connection;
+      }
+    } catch (SQLException ignore) {
+      // If isClosed throws fall through and connect again
+    }
+    return getNewConnection(config);
+  }
+
+  <K> List<ColumnValue> getColumnToValueList(ConnectionConfiguration config,
+      RegionMapping regionMapping, K key, PdxInstance value, Operation operation) {
+    String keyColumnName = getKeyColumnName(config, regionMapping.getTableName());
+    ColumnValue keyColumnValue = new ColumnValue(true, keyColumnName, key);
+
+    if (operation.isDestroy() || operation.isGet()) {
+      return Collections.singletonList(keyColumnValue);
+    }
+
+    List<ColumnValue> result = createColumnValueList(regionMapping, value, keyColumnName);
+    result.add(keyColumnValue);
+    return result;
+  }
+
+  void close() {
+    connectionMap.values().forEach(this::close);
+  }
+
+  String getKeyColumnName(ConnectionConfiguration connectionConfig, String tableName) {
+    return tableToPrimaryKeyMap.computeIfAbsent(tableName,
+        k -> computeKeyColumnName(connectionConfig, k));
+  }
+
+  ConnectionConfiguration getConnectionConfig(String connectionConfigName) {
+    return configService.getConnectionConfig(connectionConfigName);
+  }
+
+  PreparedStatement getPreparedStatement(Connection connection, List<ColumnValue> columnList,
+      String tableName, Operation operation, int pdxTypeId) {
+    PreparedStatementCache statementCache = preparedStatementCache.get();
+
+    if (statementCache == null) {
+      statementCache = new PreparedStatementCache();
+      preparedStatementCache.set(statementCache);
+    }
+
+    return statementCache.getPreparedStatement(connection, columnList, tableName, operation,
+        pdxTypeId);
+  }
+
+  // package protected for testing purposes only
+  Connection getSQLConnection(ConnectionConfiguration config) throws SQLException {
+    return DriverManager.getConnection(config.getUrl(), config.getUser(), config.getPassword());
+  }
+
+  private synchronized Connection getNewConnection(ConnectionConfiguration config) {
+    Connection connection;
+    try {
+      connection = getSQLConnection(config);
+    } catch (SQLException e) {
+      // TODO: consider a different exception
+      throw new IllegalStateException("Could not connect to " + config.getUrl(), e);
+    }
+    connectionMap.put(config.getName(), connection);
+    return connection;
+  }
+
+  private List<ColumnValue> createColumnValueList(RegionMapping regionMapping, PdxInstance value,
+      String keyColumnName) {
+    List<ColumnValue> result = new ArrayList<>();
+    for (String fieldName : value.getFieldNames()) {
+      String columnName = regionMapping.getColumnNameForField(fieldName);
+      if (columnName.equalsIgnoreCase(keyColumnName)) {
+        continue;
+      }
+      ColumnValue columnValue = new ColumnValue(false, columnName, value.getField(fieldName));
+      result.add(columnValue);
+    }
+    return result;
+  }
+
+  private String computeKeyColumnName(ConnectionConfiguration connectionConfig, String tableName) {
+    // TODO: check config for key column
+    String key = null;
+    try {
+      Connection connection = getConnection(connectionConfig);
+      DatabaseMetaData metaData = connection.getMetaData();
+      ResultSet tables = metaData.getTables(null, null, "%", null);
+
+      String realTableName = getTableNameFromMetaData(tableName, tables);
+      key = getPrimaryKeyColumnNameFromMetaData(realTableName, metaData);
+
+    } catch (SQLException e) {
+      handleSQLException(e);
+    }
+    return key;
+  }
+
+  private String getTableNameFromMetaData(String tableName, ResultSet tables) throws SQLException {
+    String realTableName = null;
+    while (tables.next()) {
+      String name = tables.getString("TABLE_NAME");
+      if (name.equalsIgnoreCase(tableName)) {
+        if (realTableName != null) {
+          throw new IllegalStateException("Duplicate tables that match region name");
+        }
+        realTableName = name;
+      }
+    }
+
+    if (realTableName == null) {
+      throw new IllegalStateException("no table was found that matches " + tableName);
+    }
+    return realTableName;
+  }
+
+  private String getPrimaryKeyColumnNameFromMetaData(String tableName, DatabaseMetaData metaData)
+      throws SQLException {
+    ResultSet primaryKeys = metaData.getPrimaryKeys(null, null, tableName);
+    if (!primaryKeys.next()) {
+      throw new IllegalStateException(
+          "The table " + tableName + " does not have a primary key column.");
+    }
+    String key = primaryKeys.getString("COLUMN_NAME");
+    if (primaryKeys.next()) {
+      throw new IllegalStateException(
+          "The table " + tableName + " has more than one primary key column.");
+    }
+    return key;
+  }
+
+  private void handleSQLException(SQLException e) {
+    throw new IllegalStateException("NYI: handleSQLException", e);
+  }
+
+  private void close(Connection connection) {
+    if (connection != null) {
+      try {
+        connection.close();
+      } catch (SQLException ignore) {
+      }
+    }
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/InternalJdbcConnectorService.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/InternalJdbcConnectorService.java
new file mode 100644
index 0000000..d0bd167
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/InternalJdbcConnectorService.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.internal.cache.CacheService;
+import org.apache.geode.internal.cache.extension.Extension;
+
+public interface InternalJdbcConnectorService extends Extension<Cache>, CacheService {
+  void addOrUpdateConnectionConfig(ConnectionConfiguration config);
+
+  void addOrUpdateRegionMapping(RegionMapping mapping);
+
+  ConnectionConfiguration getConnectionConfig(String connectionName);
+
+  RegionMapping getMappingForRegion(String regionName);
+
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/JdbcConnectorService.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/JdbcConnectorService.java
new file mode 100644
index 0000000..9af7aeb
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/JdbcConnectorService.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlGenerator;
+import org.apache.geode.internal.cache.CacheService;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.extension.Extensible;
+import org.apache.geode.internal.cache.xmlcache.XmlGenerator;
+import org.apache.geode.management.internal.beans.CacheServiceMBeanBase;
+
+public class JdbcConnectorService implements InternalJdbcConnectorService {
+
+  private final Map<String, ConnectionConfiguration> connectionsByName = new ConcurrentHashMap<>();
+  private final Map<String, RegionMapping> mappingsByRegion = new ConcurrentHashMap<>();
+  private volatile InternalCache cache;
+  private boolean registered;
+
+  public ConnectionConfiguration getConnectionConfig(String connectionName) {
+    return connectionsByName.get(connectionName);
+  }
+
+  public RegionMapping getMappingForRegion(String regionName) {
+    return mappingsByRegion.get(regionName);
+  }
+
+  @Override
+  public void addOrUpdateConnectionConfig(ConnectionConfiguration config) {
+    registerAsExtension();
+    connectionsByName.put(config.getName(), config);
+  }
+
+  @Override
+  public void addOrUpdateRegionMapping(RegionMapping mapping) {
+    registerAsExtension();
+    mappingsByRegion.put(mapping.getRegionName(), mapping);
+  }
+
+  @Override
+  public void init(Cache cache) {
+    this.cache = (InternalCache) cache;
+  }
+
+  private synchronized void registerAsExtension() {
+    if (!registered) {
+      cache.getExtensionPoint().addExtension(this);
+      registered = true;
+    }
+  }
+
+  @Override
+  public Class<? extends CacheService> getInterface() {
+    return InternalJdbcConnectorService.class;
+  }
+
+  @Override
+  public CacheServiceMBeanBase getMBean() {
+    return null;
+  }
+
+  @Override
+  public XmlGenerator<Cache> getXmlGenerator() {
+    return new JdbcConnectorServiceXmlGenerator(connectionsByName.values(),
+        mappingsByRegion.values());
+  }
+
+  @Override
+  public void beforeCreate(Extensible<Cache> source, Cache cache) {
+    // nothing
+  }
+
+  @Override
+  public void onCreate(Extensible<Cache> source, Extensible<Cache> target) {
+    // nothing
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/PreparedStatementCache.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/PreparedStatementCache.java
new file mode 100644
index 0000000..bcc6700
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/PreparedStatementCache.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.geode.cache.Operation;
+
+class PreparedStatementCache {
+
+  private SqlStatementFactory statementFactory = new SqlStatementFactory();
+  // TODO: if connection lost, we will still keep the statement. Make LRU?
+  private Map<StatementKey, PreparedStatement> statements = new HashMap<>();
+
+  PreparedStatement getPreparedStatement(Connection connection, List<ColumnValue> columnList,
+      String tableName, Operation operation, int pdxTypeId) {
+    StatementKey key = new StatementKey(pdxTypeId, operation, tableName);
+    return statements.computeIfAbsent(key, k -> {
+      String sqlStr = getSqlString(tableName, columnList, operation);
+      PreparedStatement statement = null;
+      try {
+        statement = connection.prepareStatement(sqlStr);
+      } catch (SQLException e) {
+        handleSQLException(e);
+      }
+      return statement;
+    });
+  }
+
+  private String getSqlString(String tableName, List<ColumnValue> columnList, Operation operation) {
+    if (operation.isCreate()) {
+      return statementFactory.createInsertSqlString(tableName, columnList);
+    } else if (operation.isUpdate()) {
+      return statementFactory.createUpdateSqlString(tableName, columnList);
+    } else if (operation.isDestroy()) {
+      return statementFactory.createDestroySqlString(tableName, columnList);
+    } else if (operation.isGet()) {
+      return statementFactory.createSelectQueryString(tableName, columnList);
+    } else {
+      throw new IllegalArgumentException("unsupported operation " + operation);
+    }
+  }
+
+  private void handleSQLException(SQLException e) {
+    throw new IllegalStateException("NYI: handleSQLException", e);
+  }
+
+  private static class StatementKey {
+    private final int pdxTypeId;
+    private final Operation operation;
+    private final String tableName;
+
+    StatementKey(int pdxTypeId, Operation operation, String tableName) {
+      this.pdxTypeId = pdxTypeId;
+      this.operation = operation;
+      this.tableName = tableName;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+
+      StatementKey that = (StatementKey) o;
+
+      if (pdxTypeId != that.pdxTypeId) {
+        return false;
+      }
+      if (operation != null ? !operation.equals(that.operation) : that.operation != null) {
+        return false;
+      }
+      return tableName != null ? tableName.equals(that.tableName) : that.tableName == null;
+    }
+
+    @Override
+    public int hashCode() {
+      int result = pdxTypeId;
+      result = 31 * result + (operation != null ? operation.hashCode() : 0);
+      result = 31 * result + (tableName != null ? tableName.hashCode() : 0);
+      return result;
+    }
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/RegionMapping.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/RegionMapping.java
new file mode 100644
index 0000000..327775c
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/RegionMapping.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class RegionMapping {
+  private final String regionName;
+  private final String pdxClassName;
+  private final String tableName;
+  private final String connectionConfigName;
+  private final boolean primaryKeyInValue;
+  private final Map<String, String> fieldToColumnMap;
+
+  public RegionMapping(String regionName, String pdxClassName, String tableName,
+      String connectionConfigName, boolean primaryKeyInValue,
+      Map<String, String> fieldToColumnMap) {
+    this.regionName = regionName;
+    this.pdxClassName = pdxClassName;
+    this.tableName = tableName;
+    this.connectionConfigName = connectionConfigName;
+    this.primaryKeyInValue = primaryKeyInValue;
+    this.fieldToColumnMap = fieldToColumnMap;
+  }
+
+  public String getConnectionConfigName() {
+    return connectionConfigName;
+  }
+
+  public String getRegionName() {
+    return regionName;
+  }
+
+  public String getPdxClassName() {
+    return pdxClassName;
+  }
+
+  public String getTableName() {
+    return tableName;
+  }
+
+  public boolean isPrimaryKeyInValue() {
+    return primaryKeyInValue;
+  }
+
+  public String getColumnNameForField(String fieldName) {
+    String columnName = fieldToColumnMap.get(fieldName);
+    return columnName != null ? columnName : fieldName;
+  }
+
+  public Map<String, String> getFieldToColumnMap() {
+    return Collections.unmodifiableMap(fieldToColumnMap);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    RegionMapping that = (RegionMapping) o;
+
+    if (primaryKeyInValue != that.primaryKeyInValue) {
+      return false;
+    }
+    if (regionName != null ? !regionName.equals(that.regionName) : that.regionName != null) {
+      return false;
+    }
+    if (pdxClassName != null ? !pdxClassName.equals(that.pdxClassName)
+        : that.pdxClassName != null) {
+      return false;
+    }
+    if (tableName != null ? !tableName.equals(that.tableName) : that.tableName != null) {
+      return false;
+    }
+    if (connectionConfigName != null ? !connectionConfigName.equals(that.connectionConfigName)
+        : that.connectionConfigName != null) {
+      return false;
+    }
+    return fieldToColumnMap != null ? fieldToColumnMap.equals(that.fieldToColumnMap)
+        : that.fieldToColumnMap == null;
+  }
+
+  @Override
+  public int hashCode() {
+    int result = regionName != null ? regionName.hashCode() : 0;
+    result = 31 * result + (pdxClassName != null ? pdxClassName.hashCode() : 0);
+    result = 31 * result + (tableName != null ? tableName.hashCode() : 0);
+    result = 31 * result + (connectionConfigName != null ? connectionConfigName.hashCode() : 0);
+    result = 31 * result + (primaryKeyInValue ? 1 : 0);
+    result = 31 * result + (fieldToColumnMap != null ? fieldToColumnMap.hashCode() : 0);
+    return result;
+  }
+
+  @Override
+  public String toString() {
+    return "RegionMapping{" + "regionName='" + regionName + '\'' + ", pdxClassName='" + pdxClassName
+        + '\'' + ", tableName='" + tableName + '\'' + ", connectionConfigName='"
+        + connectionConfigName + '\'' + ", primaryKeyInValue=" + primaryKeyInValue
+        + ", fieldToColumnMap=" + fieldToColumnMap + '}';
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/SqlHandler.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/SqlHandler.java
new file mode 100644
index 0000000..b73698c
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/SqlHandler.java
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.pdx.PdxInstanceFactory;
+import org.apache.geode.pdx.internal.PdxInstanceImpl;
+
+public class SqlHandler {
+  private ConnectionManager manager;
+
+  public SqlHandler(ConnectionManager manager) {
+    this.manager = manager;
+  }
+
+  public void close() {
+    manager.close();
+  }
+
+  public <K, V> PdxInstance read(Region<K, V> region, K key) {
+    if (key == null) {
+      throw new IllegalArgumentException("Key for query cannot be null");
+    }
+
+    RegionMapping regionMapping = manager.getMappingForRegion(region.getName());
+    ConnectionConfiguration connectionConfig =
+        manager.getConnectionConfig(regionMapping.getConnectionConfigName());
+
+    List<ColumnValue> columnList =
+        manager.getColumnToValueList(connectionConfig, regionMapping, key, null, Operation.GET);
+    String tableName = regionMapping.getTableName();
+    PreparedStatement statement = manager.getPreparedStatement(
+        manager.getConnection(connectionConfig), columnList, tableName, Operation.GET, 0);
+    PdxInstanceFactory factory = getPdxInstanceFactory(region, regionMapping);
+    String keyColumnName = manager.getKeyColumnName(connectionConfig, tableName);
+    return executeReadStatement(statement, columnList, factory, regionMapping, keyColumnName);
+  }
+
+  private <K, V> PdxInstanceFactory getPdxInstanceFactory(Region<K, V> region,
+      RegionMapping regionMapping) {
+    InternalCache cache = (InternalCache) region.getRegionService();
+    String valueClassName = regionMapping.getPdxClassName();
+    PdxInstanceFactory factory;
+    if (valueClassName != null) {
+      factory = cache.createPdxInstanceFactory(valueClassName);
+    } else {
+      factory = cache.createPdxInstanceFactory("no class", false);
+    }
+    return factory;
+  }
+
+  private PdxInstance executeReadStatement(PreparedStatement statement,
+      List<ColumnValue> columnList, PdxInstanceFactory factory, RegionMapping regionMapping,
+      String keyColumnName) {
+    PdxInstance pdxInstance = null;
+    synchronized (statement) {
+      try {
+        setValuesInStatement(statement, columnList);
+        ResultSet resultSet = statement.executeQuery();
+        if (resultSet.next()) {
+
+          ResultSetMetaData metaData = resultSet.getMetaData();
+          int ColumnsNumber = metaData.getColumnCount();
+          for (int i = 1; i <= ColumnsNumber; i++) {
+            Object columnValue = resultSet.getObject(i);
+            String columnName = metaData.getColumnName(i);
+            String fieldName = mapColumnNameToFieldName(columnName);
+            if (regionMapping.isPrimaryKeyInValue()
+                || !keyColumnName.equalsIgnoreCase(columnName)) {
+              factory.writeField(fieldName, columnValue, Object.class);
+            }
+          }
+          if (resultSet.next()) {
+            throw new IllegalStateException(
+                "Multiple rows returned for query: " + resultSet.getStatement().toString());
+          }
+          pdxInstance = factory.create();
+        }
+      } catch (SQLException e) {
+        handleSQLException(e);
+      } finally {
+        clearStatementParameters(statement);
+      }
+    }
+    return pdxInstance;
+  }
+
+  private void setValuesInStatement(PreparedStatement statement, List<ColumnValue> columnList)
+      throws SQLException {
+    int index = 0;
+    for (ColumnValue columnValue : columnList) {
+      index++;
+      statement.setObject(index, columnValue.getValue());
+    }
+  }
+
+  private String mapColumnNameToFieldName(String columnName) {
+    return columnName.toLowerCase();
+  }
+
+  public <K, V> void write(Region<K, V> region, Operation operation, K key, PdxInstance value) {
+    if (value == null && operation != Operation.DESTROY) {
+      throw new IllegalArgumentException("PdxInstance cannot be null for non-destroy operations");
+    }
+    RegionMapping regionMapping = manager.getMappingForRegion(region.getName());
+    final String tableName = regionMapping.getTableName();
+    ConnectionConfiguration connectionConfig =
+        manager.getConnectionConfig(regionMapping.getConnectionConfigName());
+    List<ColumnValue> columnList =
+        manager.getColumnToValueList(connectionConfig, regionMapping, key, value, operation);
+
+    int pdxTypeId = value == null ? 0 : ((PdxInstanceImpl) value).getPdxType().getTypeId();
+    PreparedStatement statement = manager.getPreparedStatement(
+        manager.getConnection(connectionConfig), columnList, tableName, operation, pdxTypeId);
+    int updateCount = executeWriteStatement(statement, columnList, operation, false);
+
+    // Destroy action not guaranteed to modify any database rows
+    if (operation.isDestroy()) {
+      return;
+    }
+
+    if (updateCount <= 0) {
+      Operation upsertOp = getOppositeOperation(operation);
+      PreparedStatement upsertStatement = manager.getPreparedStatement(
+          manager.getConnection(connectionConfig), columnList, tableName, upsertOp, pdxTypeId);
+      updateCount = executeWriteStatement(upsertStatement, columnList, upsertOp, true);
+    }
+
+    if (updateCount != 1) {
+      throw new IllegalStateException("Unexpected updateCount " + updateCount);
+    }
+  }
+
+  private Operation getOppositeOperation(Operation operation) {
+    return operation.isUpdate() ? Operation.CREATE : Operation.UPDATE;
+  }
+
+  private int executeWriteStatement(PreparedStatement statement, List<ColumnValue> columnList,
+      Operation operation, boolean handleException) {
+    int updateCount = 0;
+    synchronized (statement) {
+      try {
+        setValuesInStatement(statement, columnList);
+        updateCount = statement.executeUpdate();
+      } catch (SQLException e) {
+        if (handleException || operation.isDestroy()) {
+          handleSQLException(e);
+        }
+      } finally {
+        clearStatementParameters(statement);
+      }
+    }
+    return updateCount;
+  }
+
+  private void clearStatementParameters(PreparedStatement statement) {
+    try {
+      statement.clearParameters();
+    } catch (SQLException ignore) {
+    }
+  }
+
+  private void handleSQLException(SQLException e) {
+    throw new IllegalStateException("NYI: handleSQLException", e);
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/SqlStatementFactory.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/SqlStatementFactory.java
new file mode 100644
index 0000000..d5367ef
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/SqlStatementFactory.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import java.util.List;
+
+class SqlStatementFactory {
+
+  String createSelectQueryString(String tableName, List<ColumnValue> columnList) {
+    assert columnList.size() == 1;
+    ColumnValue keyCV = columnList.get(0);
+    assert keyCV.isKey();
+    return "SELECT * FROM " + tableName + " WHERE " + keyCV.getColumnName() + " = ?";
+  }
+
+  String createDestroySqlString(String tableName, List<ColumnValue> columnList) {
+    assert columnList.size() == 1;
+    ColumnValue keyCV = columnList.get(0);
+    assert keyCV.isKey();
+    return "DELETE FROM " + tableName + " WHERE " + keyCV.getColumnName() + " = ?";
+  }
+
+  String createUpdateSqlString(String tableName, List<ColumnValue> columnList) {
+    StringBuilder query = new StringBuilder("UPDATE " + tableName + " SET ");
+    int idx = 0;
+    for (ColumnValue column : columnList) {
+      if (!column.isKey()) {
+        idx++;
+        if (idx > 1) {
+          query.append(", ");
+        }
+        query.append(column.getColumnName());
+        query.append(" = ?");
+      }
+    }
+    for (ColumnValue column : columnList) {
+      if (column.isKey()) {
+        query.append(" WHERE ");
+        query.append(column.getColumnName());
+        query.append(" = ?");
+        // currently only support simple primary key with one column
+        break;
+      }
+    }
+    return query.toString();
+  }
+
+  String createInsertSqlString(String tableName, List<ColumnValue> columnList) {
+    StringBuilder columnNames = new StringBuilder("INSERT INTO " + tableName + " (");
+    StringBuilder columnValues = new StringBuilder(" VALUES (");
+    int columnCount = columnList.size();
+    int idx = 0;
+    for (ColumnValue column : columnList) {
+      idx++;
+      columnNames.append(column.getColumnName());
+      columnValues.append('?');
+      if (idx != columnCount) {
+        columnNames.append(", ");
+        columnValues.append(",");
+      }
+    }
+    columnNames.append(")");
+    columnValues.append(")");
+    return columnNames.append(columnValues).toString();
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/ConnectionConfigBuilder.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/ConnectionConfigBuilder.java
new file mode 100644
index 0000000..34d61ad
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/ConnectionConfigBuilder.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+
+class ConnectionConfigBuilder {
+  private String name;
+  private String url;
+  private String user;
+  private String password;
+
+  ConnectionConfigBuilder withName(String name) {
+    this.name = name;
+    return this;
+  }
+
+  ConnectionConfigBuilder withUrl(String url) {
+    this.url = url;
+    return this;
+  }
+
+  ConnectionConfigBuilder withUser(String user) {
+    this.user = user;
+    return this;
+  }
+
+  ConnectionConfigBuilder withPassword(String password) {
+    this.password = password;
+    return this;
+  }
+
+  ConnectionConfiguration build() {
+    return new ConnectionConfiguration(name, url, user, password);
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/ElementType.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/ElementType.java
new file mode 100644
index 0000000..580910c
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/ElementType.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import java.util.Stack;
+
+import org.xml.sax.Attributes;
+
+import org.apache.geode.cache.CacheXmlException;
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.internal.cache.xmlcache.CacheCreation;
+
+enum ElementType {
+  CONNECTION_SERVICE("connector-service") {
+    @Override
+    void startElement(Stack<Object> stack, Attributes attributes) {
+      if (!(stack.peek() instanceof CacheCreation)) {
+        throw new CacheXmlException(
+            "jdbc <connector-service> elements must occur within <cache> elements");
+      }
+      CacheCreation cacheCreation = (CacheCreation) stack.peek();
+      JdbcServiceConfiguration serviceConfig = new JdbcServiceConfiguration();
+      cacheCreation.getExtensionPoint().addExtension(serviceConfig);
+      stack.push(serviceConfig);
+    }
+
+    @Override
+    void endElement(Stack<Object> stack) {
+      stack.pop();
+    }
+  },
+  CONNECTION("connection") {
+    @Override
+    void startElement(Stack<Object> stack, Attributes attributes) {
+      if (!(stack.peek() instanceof JdbcServiceConfiguration)) {
+        throw new CacheXmlException(
+            "jdbc <connection> elements must occur within <connector-service> elements");
+      }
+      ConnectionConfigBuilder connectionConfig = new ConnectionConfigBuilder()
+          .withName(attributes.getValue(JdbcConnectorServiceXmlParser.NAME))
+          .withUrl(attributes.getValue(JdbcConnectorServiceXmlParser.URL))
+          .withUser(attributes.getValue(JdbcConnectorServiceXmlParser.USER))
+          .withPassword(attributes.getValue(JdbcConnectorServiceXmlParser.PASSWORD));
+      stack.push(connectionConfig);
+    }
+
+    @Override
+    void endElement(Stack<Object> stack) {
+      ConnectionConfiguration config = ((ConnectionConfigBuilder) stack.pop()).build();
+      JdbcServiceConfiguration connectorService = (JdbcServiceConfiguration) stack.peek();
+      connectorService.addConnectionConfig(config);
+    }
+  },
+  REGION_MAPPING("region-mapping") {
+    @Override
+    void startElement(Stack<Object> stack, Attributes attributes) {
+      if (!(stack.peek() instanceof JdbcServiceConfiguration)) {
+        throw new CacheXmlException(
+            "jdbc <region-mapping> elements must occur within <connector-service> elements");
+      }
+      RegionMappingBuilder mapping = new RegionMappingBuilder()
+          .withRegionName(attributes.getValue(JdbcConnectorServiceXmlParser.REGION))
+          .withConnectionConfigName(
+              attributes.getValue(JdbcConnectorServiceXmlParser.CONNECTION_NAME))
+          .withTableName(attributes.getValue(JdbcConnectorServiceXmlParser.TABLE))
+          .withPdxClassName(attributes.getValue(JdbcConnectorServiceXmlParser.PDX_CLASS))
+          .withPrimaryKeyInValue(
+              attributes.getValue(JdbcConnectorServiceXmlParser.PRIMARY_KEY_IN_VALUE));
+      stack.push(mapping);
+    }
+
+    @Override
+    void endElement(Stack<Object> stack) {
+      RegionMapping mapping = ((RegionMappingBuilder) stack.pop()).build();
+      JdbcServiceConfiguration connectorService = (JdbcServiceConfiguration) stack.peek();
+      connectorService.addRegionMapping(mapping);
+    }
+  },
+  FIELD_MAPPING("field-mapping") {
+    @Override
+    void startElement(Stack<Object> stack, Attributes attributes) {
+      if (!(stack.peek() instanceof RegionMappingBuilder)) {
+        throw new CacheXmlException(
+            "jdbc <field-mapping> elements must occur within <region-mapping> elements");
+      }
+      RegionMappingBuilder mapping = (RegionMappingBuilder) stack.peek();
+      String fieldName = attributes.getValue(JdbcConnectorServiceXmlParser.FIELD_NAME);
+      String columnName = attributes.getValue(JdbcConnectorServiceXmlParser.COLUMN_NAME);
+      mapping.withFieldToColumnMapping(fieldName, columnName);
+    }
+
+    @Override
+    void endElement(Stack<Object> stack) {}
+  };
+
+  private String typeName;
+
+  ElementType(String typeName) {
+    this.typeName = typeName;
+  }
+
+  static ElementType getTypeFromName(String typeName) {
+    for (ElementType type : ElementType.values()) {
+      if (type.typeName.equals(typeName))
+        return type;
+    }
+    throw new IllegalArgumentException("Invalid type '" + typeName + "'");
+  }
+
+  String getTypeName() {
+    return typeName;
+  }
+
+  abstract void startElement(Stack<Object> stack, Attributes attributes);
+
+  abstract void endElement(Stack<Object> stack);
+
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGenerator.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGenerator.java
new file mode 100644
index 0000000..867dd7f
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGenerator.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.COLUMN_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.CONNECTION_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.FIELD_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.NAMESPACE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PASSWORD;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PDX_CLASS;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PRIMARY_KEY_IN_VALUE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.REGION;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.TABLE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.URL;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.USER;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+
+import org.xml.sax.ContentHandler;
+import org.xml.sax.SAXException;
+import org.xml.sax.helpers.AttributesImpl;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.internal.cache.xmlcache.CacheXmlGenerator;
+import org.apache.geode.internal.cache.xmlcache.XmlGenerator;
+import org.apache.geode.internal.cache.xmlcache.XmlGeneratorUtils;
+
+public class JdbcConnectorServiceXmlGenerator implements XmlGenerator<Cache> {
+  private static final AttributesImpl EMPTY = new AttributesImpl();
+  static final String PREFIX = "jdbc";
+
+  private final Collection<ConnectionConfiguration> connections;
+  private final Collection<RegionMapping> mappings;
+
+  public JdbcConnectorServiceXmlGenerator(Collection<ConnectionConfiguration> connections,
+      Collection<RegionMapping> mappings) {
+    this.connections = connections != null ? connections : Collections.emptyList();
+    this.mappings = mappings != null ? mappings : Collections.emptyList();
+  }
+
+  @Override
+  public String getNamespaceUri() {
+    return NAMESPACE;
+  }
+
+  @Override
+  public void generate(CacheXmlGenerator cacheXmlGenerator) throws SAXException {
+    final ContentHandler handler = cacheXmlGenerator.getContentHandler();
+
+    handler.startPrefixMapping(PREFIX, NAMESPACE);
+    XmlGeneratorUtils.startElement(handler, PREFIX, ElementType.CONNECTION_SERVICE.getTypeName(),
+        EMPTY);
+    for (ConnectionConfiguration connection : connections) {
+      outputConnectionConfiguration(handler, connection);
+    }
+    for (RegionMapping mapping : mappings) {
+      outputRegionMapping(handler, mapping);
+    }
+    XmlGeneratorUtils.endElement(handler, PREFIX, ElementType.CONNECTION_SERVICE.getTypeName());
+  }
+
+  /**
+   * For testing only
+   */
+  Collection<ConnectionConfiguration> getConnections() {
+    return connections;
+  }
+
+  /**
+   * For testing only
+   */
+  Collection<RegionMapping> getMappings() {
+    return mappings;
+  }
+
+  private void outputConnectionConfiguration(ContentHandler handler, ConnectionConfiguration config)
+      throws SAXException {
+    AttributesImpl attributes = new AttributesImpl();
+    XmlGeneratorUtils.addAttribute(attributes, NAME, config.getName());
+    XmlGeneratorUtils.addAttribute(attributes, URL, config.getUrl());
+    XmlGeneratorUtils.addAttribute(attributes, USER, config.getUser());
+    XmlGeneratorUtils.addAttribute(attributes, PASSWORD, config.getPassword());
+    XmlGeneratorUtils.emptyElement(handler, PREFIX, ElementType.CONNECTION.getTypeName(),
+        attributes);
+  }
+
+  private void outputRegionMapping(ContentHandler handler, RegionMapping mapping)
+      throws SAXException {
+    AttributesImpl attributes = new AttributesImpl();
+    XmlGeneratorUtils.addAttribute(attributes, CONNECTION_NAME, mapping.getConnectionConfigName());
+    XmlGeneratorUtils.addAttribute(attributes, REGION, mapping.getRegionName());
+    XmlGeneratorUtils.addAttribute(attributes, TABLE, mapping.getTableName());
+    XmlGeneratorUtils.addAttribute(attributes, PDX_CLASS, mapping.getPdxClassName());
+    XmlGeneratorUtils.addAttribute(attributes, PRIMARY_KEY_IN_VALUE,
+        Boolean.toString(mapping.isPrimaryKeyInValue()));
+
+    XmlGeneratorUtils.startElement(handler, PREFIX, ElementType.REGION_MAPPING.getTypeName(),
+        attributes);
+    addFieldMappings(handler, mapping.getFieldToColumnMap());
+    XmlGeneratorUtils.endElement(handler, PREFIX, ElementType.REGION_MAPPING.getTypeName());
+  }
+
+  private void addFieldMappings(ContentHandler handler, Map<String, String> fieldMappings)
+      throws SAXException {
+    for (Map.Entry<String, String> fieldMapping : fieldMappings.entrySet()) {
+      AttributesImpl fieldAttributes = new AttributesImpl();
+      XmlGeneratorUtils.addAttribute(fieldAttributes, FIELD_NAME, fieldMapping.getKey());
+      XmlGeneratorUtils.addAttribute(fieldAttributes, COLUMN_NAME, fieldMapping.getValue());
+      XmlGeneratorUtils.emptyElement(handler, PREFIX, ElementType.FIELD_MAPPING.getTypeName(),
+          fieldAttributes);
+    }
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlParser.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlParser.java
new file mode 100644
index 0000000..7caaa1d
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlParser.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import org.xml.sax.Attributes;
+import org.xml.sax.SAXException;
+
+import org.apache.geode.internal.cache.xmlcache.AbstractXmlParser;
+
+public class JdbcConnectorServiceXmlParser extends AbstractXmlParser {
+  static final String NAMESPACE = "http://geode.apache.org/schema/jdbc-connector";
+  static final String NAME = "name";
+  static final String URL = "url";
+  static final String USER = "user";
+  static final String PASSWORD = "password";
+  static final String REGION = "region";
+  static final String CONNECTION_NAME = "connection-name";
+  static final String TABLE = "table";
+  static final String PDX_CLASS = "pdx-class";
+  static final String FIELD_NAME = "field-name";
+  static final String COLUMN_NAME = "column-name";
+  static final String PRIMARY_KEY_IN_VALUE = "primary-key-in-value";
+
+  @Override
+  public String getNamespaceUri() {
+    return NAMESPACE;
+  }
+
+  @Override
+  public void startElement(String uri, String localName, String qName, Attributes attributes)
+      throws SAXException {
+    if (!NAMESPACE.equals(uri)) {
+      return;
+    }
+    ElementType.getTypeFromName(localName).startElement(stack, attributes);
+  }
+
+  @Override
+  public void endElement(String uri, String localName, String qName) throws SAXException {
+    if (!NAMESPACE.equals(uri)) {
+      return;
+    }
+    ElementType.getTypeFromName(localName).endElement(stack);
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcServiceConfiguration.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcServiceConfiguration.java
new file mode 100644
index 0000000..2892334
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcServiceConfiguration.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.connectors.jdbc.internal.InternalJdbcConnectorService;
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.extension.Extensible;
+import org.apache.geode.internal.cache.extension.Extension;
+import org.apache.geode.internal.cache.xmlcache.XmlGenerator;
+
+public class JdbcServiceConfiguration implements Extension<Cache> {
+
+  private final List<ConnectionConfiguration> connections = new ArrayList<>();
+  private final List<RegionMapping> mappings = new ArrayList<>();
+
+  void addConnectionConfig(ConnectionConfiguration config) {
+    connections.add(config);
+  }
+
+  void addRegionMapping(RegionMapping mapping) {
+    mappings.add(mapping);
+  }
+
+  @Override
+  public XmlGenerator<Cache> getXmlGenerator() {
+    return new JdbcConnectorServiceXmlGenerator(connections, mappings);
+  }
+
+  @Override
+  public void beforeCreate(Extensible<Cache> source, Cache cache) {
+    // nothing
+  }
+
+  @Override
+  public void onCreate(Extensible<Cache> source, Extensible<Cache> target) {
+    InternalCache internalCache = (InternalCache) target;
+    InternalJdbcConnectorService service =
+        internalCache.getService(InternalJdbcConnectorService.class);
+    connections.forEach(connection -> service.addOrUpdateConnectionConfig(connection));
+    mappings.forEach(mapping -> service.addOrUpdateRegionMapping(mapping));
+  }
+}
diff --git a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/RegionMappingBuilder.java b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/RegionMappingBuilder.java
new file mode 100644
index 0000000..b4adcaf
--- /dev/null
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/internal/xml/RegionMappingBuilder.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+
+class RegionMappingBuilder {
+
+  private String regionName;
+  private String pdxClassName;
+  private String tableName;
+  private String connectionConfigName;
+  private boolean primaryKeyInValue;
+  private final Map<String, String> fieldToColumnMap = new HashMap<>();
+
+  RegionMappingBuilder withRegionName(String regionName) {
+    this.regionName = regionName;
+    return this;
+  }
+
+  RegionMappingBuilder withPdxClassName(String pdxClassName) {
+    this.pdxClassName = pdxClassName;
+    return this;
+  }
+
+  RegionMappingBuilder withTableName(String tableName) {
+    this.tableName = tableName;
+    return this;
+  }
+
+  RegionMappingBuilder withConnectionConfigName(String connectionConfigName) {
+    this.connectionConfigName = connectionConfigName;
+    return this;
+  }
+
+  RegionMappingBuilder withPrimaryKeyInValue(String primaryKeyInValue) {
+    this.primaryKeyInValue = Boolean.parseBoolean(primaryKeyInValue);
+    return this;
+  }
+
+  RegionMappingBuilder withFieldToColumnMapping(String fieldName, String columnMapping) {
+    this.fieldToColumnMap.put(fieldName, columnMapping);
+    return this;
+  }
+
+  RegionMapping build() {
+    return new RegionMapping(regionName, pdxClassName, tableName, connectionConfigName,
+        primaryKeyInValue, fieldToColumnMap);
+  }
+}
diff --git a/geode-connectors/src/main/resources/META-INF/services/org.apache.geode.internal.cache.CacheService b/geode-connectors/src/main/resources/META-INF/services/org.apache.geode.internal.cache.CacheService
new file mode 100644
index 0000000..bc91015
--- /dev/null
+++ b/geode-connectors/src/main/resources/META-INF/services/org.apache.geode.internal.cache.CacheService
@@ -0,0 +1 @@
+org.apache.geode.connectors.jdbc.internal.JdbcConnectorService
\ No newline at end of file
diff --git a/geode-connectors/src/main/resources/META-INF/services/org.apache.geode.internal.cache.xmlcache.XmlParser b/geode-connectors/src/main/resources/META-INF/services/org.apache.geode.internal.cache.xmlcache.XmlParser
new file mode 100644
index 0000000..9fbc42a
--- /dev/null
+++ b/geode-connectors/src/main/resources/META-INF/services/org.apache.geode.internal.cache.xmlcache.XmlParser
@@ -0,0 +1 @@
+org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser
\ No newline at end of file
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/Employee.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/Employee.java
new file mode 100644
index 0000000..82e2e90
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/Employee.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import org.apache.geode.pdx.PdxReader;
+import org.apache.geode.pdx.PdxSerializable;
+import org.apache.geode.pdx.PdxWriter;
+
+@SuppressWarnings("unused")
+public class Employee implements PdxSerializable {
+  private String name;
+  private int age;
+
+  public Employee() {
+    // nothing
+  }
+
+  Employee(String name, int age) {
+    this.name = name;
+    this.age = age;
+  }
+
+  String getName() {
+    return name;
+  }
+
+  int getAge() {
+    return age;
+  }
+
+  @Override
+  public void toData(PdxWriter writer) {
+    writer.writeString("name", this.name);
+    writer.writeInt("age", this.age);
+  }
+
+  @Override
+  public void fromData(PdxReader reader) {
+    this.name = reader.readString("name");
+    this.age = reader.readInt("age");
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriterIntegrationTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriterIntegrationTest.java
new file mode 100644
index 0000000..1655bd9
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriterIntegrationTest.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import static org.apache.geode.cache.RegionShortcut.REPLICATE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.connectors.jdbc.internal.TestConfigService;
+import org.apache.geode.connectors.jdbc.internal.TestableConnectionManager;
+import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+
+@Category(IntegrationTest.class)
+public class JdbcAsyncWriterIntegrationTest {
+
+  private static final String DB_NAME = "DerbyDB";
+  private static final String REGION_TABLE_NAME = "employees";
+  private static final String CONNECTION_URL = "jdbc:derby:memory:" + DB_NAME + ";create=true";
+
+  private Cache cache;
+  private Region<String, PdxInstance> employees;
+  private Connection connection;
+  private Statement statement;
+  private JdbcAsyncWriter jdbcWriter;
+  private PdxInstance pdxEmployee1;
+  private PdxInstance pdxEmployee2;
+  private Employee employee1;
+  private Employee employee2;
+
+  @Before
+  public void setup() throws Exception {
+    cache = new CacheFactory().setPdxReadSerialized(false).create();
+    employees = createRegionWithJDBCAsyncWriter(REGION_TABLE_NAME);
+    connection = DriverManager.getConnection(CONNECTION_URL);
+    statement = connection.createStatement();
+    statement.execute("Create Table " + REGION_TABLE_NAME
+        + " (id varchar(10) primary key not null, name varchar(10), age int)");
+    pdxEmployee1 = cache.createPdxInstanceFactory(Employee.class.getName())
+        .writeString("name", "Emp1").writeInt("age", 55).create();
+    pdxEmployee2 = cache.createPdxInstanceFactory(Employee.class.getName())
+        .writeString("name", "Emp2").writeInt("age", 21).create();
+    employee1 = (Employee) pdxEmployee1.getObject();
+    employee2 = (Employee) pdxEmployee2.getObject();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    cache.close();
+    closeDB();
+  }
+
+  private void closeDB() throws Exception {
+    if (statement == null) {
+      statement = connection.createStatement();
+    }
+    statement.execute("Drop table " + REGION_TABLE_NAME);
+    statement.close();
+
+    if (connection != null) {
+      connection.close();
+    }
+  }
+
+  @Test
+  public void validateJDBCAsyncWriterTotalEvents() {
+    employees.put("1", pdxEmployee1);
+    employees.put("2", pdxEmployee2);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getTotalEvents()).isEqualTo(2));
+  }
+
+  @Test
+  public void canInsertIntoTable() throws Exception {
+    employees.put("1", pdxEmployee1);
+    employees.put("2", pdxEmployee2);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee1);
+    assertRecordMatchesEmployee(resultSet, "2", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void verifyThatPdxFieldNamedSameAsPrimaryKeyIsIgnored() throws Exception {
+    PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
+        .writeInt("age", 55).writeInt("id", 3).create();
+    employees.put("1", pdx1);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee1);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void putNonPdxInstanceFails() {
+    Region nonPdxEmployees = this.employees;
+    nonPdxEmployees.put("1", "non pdx instance");
+
+    awaitUntil(() -> assertThat(jdbcWriter.getTotalEvents()).isEqualTo(1));
+
+    assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(0);
+  }
+
+  @Test
+  public void putNonPdxInstanceThatIsPdxSerializable() throws SQLException {
+    Region nonPdxEmployees = this.employees;
+    Employee value = new Employee("Emp2", 22);
+    nonPdxEmployees.put("2", value);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "2", value);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canDestroyFromTable() throws Exception {
+    employees.put("1", pdxEmployee1);
+    employees.put("2", pdxEmployee2);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
+
+    employees.destroy("1");
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(3));
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "2", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canUpdateTable() throws Exception {
+    employees.put("1", pdxEmployee1);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
+
+    employees.put("1", pdxEmployee2);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canUpdateBecomeInsert() throws Exception {
+    employees.put("1", pdxEmployee1);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
+
+    statement.execute("delete from " + REGION_TABLE_NAME + " where id = '1'");
+    validateTableRowCount(0);
+
+    employees.put("1", pdxEmployee2);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canInsertBecomeUpdate() throws Exception {
+    statement.execute("Insert into " + REGION_TABLE_NAME + " values('1', 'bogus', 11)");
+    validateTableRowCount(1);
+
+    employees.put("1", pdxEmployee1);
+
+    awaitUntil(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee1);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  private void awaitUntil(final Runnable supplier) {
+    Awaitility.await().atMost(30, TimeUnit.SECONDS).until(supplier);
+  }
+
+  private void assertRecordMatchesEmployee(ResultSet resultSet, String key, Employee employee)
+      throws SQLException {
+    assertThat(resultSet.next()).isTrue();
+    assertThat(resultSet.getString("id")).isEqualTo(key);
+    assertThat(resultSet.getString("name")).isEqualTo(employee.getName());
+    assertThat(resultSet.getObject("age")).isEqualTo(employee.getAge());
+  }
+
+  private Region<String, PdxInstance> createRegionWithJDBCAsyncWriter(String regionName) {
+    jdbcWriter = new JdbcAsyncWriter(createSqlHandler());
+    cache.createAsyncEventQueueFactory().setBatchSize(1).setBatchTimeInterval(1)
+        .create("jdbcAsyncQueue", jdbcWriter);
+
+    RegionFactory<String, PdxInstance> regionFactory = cache.createRegionFactory(REPLICATE);
+    regionFactory.addAsyncEventQueueId("jdbcAsyncQueue");
+    return regionFactory.create(regionName);
+  }
+
+  private void validateTableRowCount(int expected) throws Exception {
+    ResultSet resultSet = statement.executeQuery("select count(*) from " + REGION_TABLE_NAME);
+    resultSet.next();
+    int size = resultSet.getInt(1);
+    assertThat(size).isEqualTo(expected);
+  }
+
+  private SqlHandler createSqlHandler() {
+    return new SqlHandler(new TestableConnectionManager(TestConfigService.getTestConfigService()));
+  }
+
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriterTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriterTest.java
new file mode 100644
index 0000000..ef285dd
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcAsyncWriterTest.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.asyncqueue.AsyncEvent;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class JdbcAsyncWriterTest {
+
+  private SqlHandler sqlHandler;
+  private JdbcAsyncWriter writer;
+
+  @Before
+  public void setup() {
+    sqlHandler = mock(SqlHandler.class);
+    writer = new JdbcAsyncWriter(sqlHandler);
+  }
+
+  @Test
+  public void throwsNullPointerExceptionIfGivenNullList() {
+    assertThatThrownBy(() -> writer.processEvents(null)).isInstanceOf(NullPointerException.class);
+  }
+
+  @Test
+  public void doesNothingIfEventListIsEmpty() {
+    writer.processEvents(Collections.emptyList());
+
+    verifyZeroInteractions(sqlHandler);
+    assertThat(writer.getSuccessfulEvents()).isZero();
+    assertThat(writer.getTotalEvents()).isZero();
+  }
+
+  @Test
+  public void writesAProvidedEvent() {
+    writer.processEvents(Collections.singletonList(createMockEvent()));
+
+    verify(sqlHandler, times(1)).write(any(), any(), any(), any());
+    assertThat(writer.getSuccessfulEvents()).isEqualTo(1);
+    assertThat(writer.getTotalEvents()).isEqualTo(1);
+  }
+
+  @Test
+  public void writesMultipleProvidedEvents() {
+    List<AsyncEvent> events = new ArrayList<>();
+    events.add(createMockEvent());
+    events.add(createMockEvent());
+    events.add(createMockEvent());
+
+    writer.processEvents(events);
+
+    verify(sqlHandler, times(3)).write(any(), any(), any(), any());
+    assertThat(writer.getSuccessfulEvents()).isEqualTo(3);
+    assertThat(writer.getTotalEvents()).isEqualTo(3);
+  }
+
+  private AsyncEvent createMockEvent() {
+    AsyncEvent event = mock(AsyncEvent.class);
+    when(event.getRegion()).thenReturn(mock(InternalRegion.class));
+    return event;
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcLoaderIntegrationTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcLoaderIntegrationTest.java
new file mode 100644
index 0000000..3905531
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcLoaderIntegrationTest.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.connectors.jdbc.internal.TestConfigService;
+import org.apache.geode.connectors.jdbc.internal.TestableConnectionManager;
+import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+
+@Category(IntegrationTest.class)
+public class JdbcLoaderIntegrationTest {
+
+  private static final String DB_NAME = "DerbyDB";
+  private static final String REGION_TABLE_NAME = "employees";
+  private static final String CONNECTION_URL = "jdbc:derby:memory:" + DB_NAME + ";create=true";
+
+  private Cache cache;
+  private Connection connection;
+  private Statement statement;
+
+  @Before
+  public void setup() throws Exception {
+    cache = new CacheFactory().setPdxReadSerialized(false).create();
+    connection = DriverManager.getConnection(CONNECTION_URL);
+    statement = connection.createStatement();
+    statement.execute("Create Table " + REGION_TABLE_NAME
+        + " (id varchar(10) primary key not null, name varchar(10), age int)");
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    cache.close();
+    closeDB();
+  }
+
+  private void closeDB() throws Exception {
+    if (statement == null) {
+      statement = connection.createStatement();
+    }
+    statement.execute("Drop table " + REGION_TABLE_NAME);
+    statement.close();
+
+    if (connection != null) {
+      connection.close();
+    }
+  }
+
+  @Test
+  public void verifySimpleGet() throws SQLException {
+    statement.execute("Insert into " + REGION_TABLE_NAME + " values('1', 'Emp1', 21)");
+    Region<String, PdxInstance> region = createRegionWithJDBCLoader(REGION_TABLE_NAME);
+    PdxInstance pdx = region.get("1");
+
+    assertThat(pdx.getField("name")).isEqualTo("Emp1");
+    assertThat(pdx.getField("age")).isEqualTo(21);
+  }
+
+  @Test
+  public void verifySimpleMiss() throws SQLException {
+    Region<String, PdxInstance> region = createRegionWithJDBCLoader(REGION_TABLE_NAME);
+    PdxInstance pdx = region.get("1");
+    assertThat(pdx).isNull();
+  }
+
+  private SqlHandler createSqlHandler() {
+    return new SqlHandler(new TestableConnectionManager(TestConfigService.getTestConfigService()));
+  }
+
+  private Region<String, PdxInstance> createRegionWithJDBCLoader(String regionName) {
+    JdbcLoader<String, PdxInstance> jdbcLoader = new JdbcLoader<>(createSqlHandler());
+    RegionFactory<String, PdxInstance> rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
+    rf.setCacheLoader(jdbcLoader);
+    return rf.create(regionName);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcLoaderTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcLoaderTest.java
new file mode 100644
index 0000000..8b08587
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcLoaderTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.LoaderHelper;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class JdbcLoaderTest {
+
+  @Test
+  public void loadReadsFromSqlHandler() {
+    SqlHandler sqlHandler = mock(SqlHandler.class);
+    JdbcLoader<Object, Object> loader = new JdbcLoader<>(sqlHandler);
+    LoaderHelper loaderHelper = mock(LoaderHelper.class);
+    when(loaderHelper.getRegion()).thenReturn(mock(InternalRegion.class));
+    loader.load(loaderHelper);
+    verify(sqlHandler, times(1)).read(any(), any());
+  }
+
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcWriterIntegrationTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcWriterIntegrationTest.java
new file mode 100644
index 0000000..e26b3ee
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcWriterIntegrationTest.java
@@ -0,0 +1,234 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import static com.googlecode.catchexception.CatchException.catchException;
+import static com.googlecode.catchexception.CatchException.caughtException;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.connectors.jdbc.internal.TestConfigService;
+import org.apache.geode.connectors.jdbc.internal.TestableConnectionManager;
+import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+
+@Category(IntegrationTest.class)
+public class JdbcWriterIntegrationTest {
+
+  private static final String DB_NAME = "DerbyDB";
+  private static final String REGION_TABLE_NAME = "employees";
+  private static final String CONNECTION_URL = "jdbc:derby:memory:" + DB_NAME + ";create=true";
+
+  private Cache cache;
+  private Region<String, PdxInstance> employees;
+  private Connection connection;
+  private Statement statement;
+  private JdbcWriter jdbcWriter;
+  private PdxInstance pdx1;
+  private PdxInstance pdx2;
+  private Employee employee1;
+  private Employee employee2;
+
+  @Before
+  public void setup() throws Exception {
+    cache = new CacheFactory().setPdxReadSerialized(false).create();
+    employees = createRegionWithJDBCSynchronousWriter(REGION_TABLE_NAME);
+    connection = DriverManager.getConnection(CONNECTION_URL);
+    statement = connection.createStatement();
+    statement.execute("Create Table " + REGION_TABLE_NAME
+        + " (id varchar(10) primary key not null, name varchar(10), age int)");
+    pdx1 = cache.createPdxInstanceFactory(Employee.class.getName()).writeString("name", "Emp1")
+        .writeInt("age", 55).create();
+    pdx2 = cache.createPdxInstanceFactory(Employee.class.getName()).writeString("name", "Emp2")
+        .writeInt("age", 21).create();
+    employee1 = (Employee) pdx1.getObject();
+    employee2 = (Employee) pdx2.getObject();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    cache.close();
+    closeDB();
+  }
+
+  private void closeDB() throws Exception {
+    if (statement == null) {
+      statement = connection.createStatement();
+    }
+    statement.execute("Drop table " + REGION_TABLE_NAME);
+    statement.close();
+
+    if (connection != null) {
+      connection.close();
+    }
+  }
+
+  @Test
+  public void canInsertIntoTable() throws Exception {
+    employees.put("1", pdx1);
+    employees.put("2", pdx2);
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee1);
+    assertRecordMatchesEmployee(resultSet, "2", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canPutAllInsertIntoTable() throws Exception {
+    Map<String, PdxInstance> putAllMap = new HashMap<>();
+    putAllMap.put("1", pdx1);
+    putAllMap.put("2", pdx2);
+    employees.putAll(putAllMap);
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee1);
+    assertRecordMatchesEmployee(resultSet, "2", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void verifyThatPdxFieldNamedSameAsPrimaryKeyIsIgnored() throws Exception {
+    PdxInstance pdxInstanceWithId = cache.createPdxInstanceFactory(Employee.class.getName())
+        .writeString("name", "Emp1").writeInt("age", 55).writeInt("id", 3).create();
+    employees.put("1", pdxInstanceWithId);
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", (Employee) pdxInstanceWithId.getObject());
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void putNonPdxInstanceFails() {
+    Region nonPdxEmployees = this.employees;
+    catchException(nonPdxEmployees).put("1", "non pdx instance");
+    assertThat((Exception) caughtException()).isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void putNonPdxInstanceThatIsPdxSerializable() throws SQLException {
+    Region nonPdxEmployees = this.employees;
+    Employee value = new Employee("Emp2", 22);
+    nonPdxEmployees.put("2", value);
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "2", value);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canDestroyFromTable() throws Exception {
+    employees.put("1", pdx1);
+    employees.put("2", pdx2);
+
+    employees.destroy("1");
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "2", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canUpdateTable() throws Exception {
+    employees.put("1", pdx1);
+    employees.put("1", pdx2);
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canUpdateBecomeInsert() throws Exception {
+    employees.put("1", pdx1);
+
+    statement.execute("delete from " + REGION_TABLE_NAME + " where id = '1'");
+    validateTableRowCount(0);
+
+    employees.put("1", pdx2);
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee2);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void canInsertBecomeUpdate() throws Exception {
+    statement.execute("Insert into " + REGION_TABLE_NAME + " values('1', 'bogus', 11)");
+    validateTableRowCount(1);
+
+    employees.put("1", pdx1);
+
+    ResultSet resultSet =
+        statement.executeQuery("select * from " + REGION_TABLE_NAME + " order by id asc");
+    assertRecordMatchesEmployee(resultSet, "1", employee1);
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  private Region<String, PdxInstance> createRegionWithJDBCSynchronousWriter(String regionName) {
+    jdbcWriter = new JdbcWriter(createSqlHandler());
+    jdbcWriter.init(new Properties());
+
+    RegionFactory<String, PdxInstance> regionFactory =
+        cache.createRegionFactory(RegionShortcut.REPLICATE);
+    regionFactory.setCacheWriter(jdbcWriter);
+    return regionFactory.create(regionName);
+  }
+
+  private void validateTableRowCount(int expected) throws Exception {
+    ResultSet resultSet = statement.executeQuery("select count(*) from " + REGION_TABLE_NAME);
+    resultSet.next();
+    int size = resultSet.getInt(1);
+    assertThat(size).isEqualTo(expected);
+  }
+
+  private SqlHandler createSqlHandler() {
+    return new SqlHandler(new TestableConnectionManager(TestConfigService.getTestConfigService()));
+  }
+
+  private void assertRecordMatchesEmployee(ResultSet resultSet, String key, Employee employee)
+      throws SQLException {
+    assertThat(resultSet.next()).isTrue();
+    assertThat(resultSet.getString("id")).isEqualTo(key);
+    assertThat(resultSet.getString("name")).isEqualTo(employee.getName());
+    assertThat(resultSet.getObject("age")).isEqualTo(employee.getAge());
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcWriterTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcWriterTest.java
new file mode 100644
index 0000000..1767c63
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JdbcWriterTest.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.RegionEvent;
+import org.apache.geode.cache.SerializedCacheValue;
+import org.apache.geode.connectors.jdbc.internal.SqlHandler;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class JdbcWriterTest {
+
+  private EntryEvent<Object, Object> entryEvent;
+  private PdxInstance pdxInstance;
+  private SqlHandler sqlHandler;
+
+  @Before
+  public void setUp() {
+    entryEvent = mock(EntryEvent.class);
+    pdxInstance = mock(PdxInstance.class);
+    SerializedCacheValue<Object> serializedNewValue = mock(SerializedCacheValue.class);
+    sqlHandler = mock(SqlHandler.class);
+
+    when(entryEvent.getRegion()).thenReturn(mock(InternalRegion.class));
+    when(entryEvent.getSerializedNewValue()).thenReturn(serializedNewValue);
+    when(serializedNewValue.getDeserializedValue()).thenReturn(pdxInstance);
+
+  }
+
+  @Test
+  public void beforeUpdateWithPdxInstanceWritesToSqlHandler() {
+    JdbcWriter<Object, Object> writer = new JdbcWriter<>(sqlHandler);
+
+    writer.beforeUpdate(entryEvent);
+
+    verify(sqlHandler, times(1)).write(any(), any(), any(), eq(pdxInstance));
+  }
+
+  @Test
+  public void beforeUpdateWithoutPdxInstanceWritesToSqlHandler() {
+    EntryEvent<Object, Object> entryEvent = mock(EntryEvent.class);
+    Object value = new Object();
+    SerializedCacheValue<Object> serializedNewValue = mock(SerializedCacheValue.class);
+    SqlHandler sqlHander = mock(SqlHandler.class);
+
+    when(entryEvent.getRegion()).thenReturn(mock(InternalRegion.class));
+    when(entryEvent.getSerializedNewValue()).thenReturn(serializedNewValue);
+    when(serializedNewValue.getDeserializedValue()).thenReturn(value);
+
+    JdbcWriter<Object, Object> writer = new JdbcWriter<>(sqlHander);
+
+    assertThatThrownBy(() -> writer.beforeUpdate(entryEvent))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void beforeCreateWithPdxInstanceWritesToSqlHandler() {
+    JdbcWriter<Object, Object> writer = new JdbcWriter<>(sqlHandler);
+
+    writer.beforeCreate(entryEvent);
+
+    verify(sqlHandler, times(1)).write(any(), any(), any(), eq(pdxInstance));
+  }
+
+  @Test
+  public void beforeDestroyWithPdxInstanceWritesToSqlHandler() {
+    JdbcWriter<Object, Object> writer = new JdbcWriter<>(sqlHandler);
+
+    writer.beforeDestroy(entryEvent);
+
+    verify(sqlHandler, times(1)).write(any(), any(), any(), eq(pdxInstance));
+  }
+
+  @Test
+  public void beforeRegionDestroyDoesNotWriteToSqlHandler() {
+    JdbcWriter<Object, Object> writer = new JdbcWriter<>(sqlHandler);
+
+    writer.beforeRegionDestroy(mock(RegionEvent.class));
+
+    verifyZeroInteractions(sqlHandler);
+  }
+
+  @Test
+  public void beforeRegionClearDoesNotWriteToSqlHandler() {
+    JdbcWriter<Object, Object> writer = new JdbcWriter<>(sqlHandler);
+
+    writer.beforeRegionClear(mock(RegionEvent.class));
+
+    verifyZeroInteractions(sqlHandler);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/AbstractJdbcCallbackTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/AbstractJdbcCallbackTest.java
new file mode 100644
index 0000000..2199ebc
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/AbstractJdbcCallbackTest.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class AbstractJdbcCallbackTest {
+
+  private AbstractJdbcCallback jdbcCallback;
+  private SqlHandler sqlHandler;
+
+  @Before
+  public void setUp() throws Exception {
+    sqlHandler = mock(SqlHandler.class);
+    jdbcCallback = new AbstractJdbcCallback(sqlHandler) {};
+  }
+
+  @Test
+  public void closesSqlHandler() throws Exception {
+    jdbcCallback.close();
+    verify(sqlHandler, times(1)).close();
+  }
+
+  @Test
+  public void returnsCorrectSqlHander() throws Exception {
+    assertThat(jdbcCallback.getSqlHandler()).isSameAs(sqlHandler);
+  }
+
+  @Test
+  public void checkInitializedDoesNothingIfInitialized() {
+    jdbcCallback.checkInitialized(mock(InternalCache.class));
+    assertThat(jdbcCallback.getSqlHandler()).isSameAs(sqlHandler);
+  }
+
+  @Test
+  public void initializedSqlHandlerIfNoneExists() {
+    jdbcCallback = new AbstractJdbcCallback() {};
+    InternalCache cache = mock(InternalCache.class);
+    InternalJdbcConnectorService service = mock(InternalJdbcConnectorService.class);
+    when(cache.getService(any())).thenReturn(service);
+    assertThat(jdbcCallback.getSqlHandler()).isNull();
+
+    jdbcCallback.checkInitialized(cache);
+
+    assertThat(jdbcCallback.getSqlHandler()).isNotNull();
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ColumnValueTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ColumnValueTest.java
new file mode 100644
index 0000000..a7468fa
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ColumnValueTest.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class ColumnValueTest {
+  private static final String COLUMN_NAME = "columnName";
+  private static final Object VALUE = new Object();
+
+  private ColumnValue value;
+
+  @Before
+  public void setup() {
+    value = new ColumnValue(true, COLUMN_NAME, VALUE);
+  }
+
+  @Test
+  public void isKeyReturnsCorrectValue() {
+    assertThat(value.isKey()).isTrue();
+  }
+
+  @Test
+  public void hasCorrectColumnName() {
+    assertThat(value.getColumnName()).isEqualTo(COLUMN_NAME);
+  }
+
+  @Test
+  public void hasCorrectValue() {
+    assertThat(value.getValue()).isSameAs(VALUE);
+  }
+
+  @Test
+  public void toStringContainsAllVariables() {
+    assertThat(value.toString()).contains(Boolean.toString(true)).contains(COLUMN_NAME)
+        .contains(VALUE.toString());
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ConnectionConfigurationTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ConnectionConfigurationTest.java
new file mode 100644
index 0000000..639e278
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ConnectionConfigurationTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class ConnectionConfigurationTest {
+
+  @Test
+  public void initiatedWithNullValues() {
+    ConnectionConfiguration config = new ConnectionConfiguration(null, null, null, null);
+    assertThat(config.getName()).isNull();
+    assertThat(config.getUrl()).isNull();
+    assertThat(config.getUser()).isNull();
+    assertThat(config.getPassword()).isNull();
+  }
+
+  @Test
+  public void hasCorrectName() {
+    String name = "name";
+    ConnectionConfiguration config = new ConnectionConfiguration(name, null, null, null);
+    assertThat(config.getName()).isEqualTo(name);
+  }
+
+  @Test
+  public void hasCorrectUrl() {
+    String url = "url";
+    ConnectionConfiguration config = new ConnectionConfiguration(null, url, null, null);
+    assertThat(config.getUrl()).isEqualTo(url);
+  }
+
+  @Test
+  public void hasCorrectUser() {
+    String user = "user";
+    ConnectionConfiguration config = new ConnectionConfiguration(null, null, user, null);
+    assertThat(config.getUser()).isEqualTo(user);
+  }
+
+  @Test
+  public void hasCorrectPassword() {
+    String password = "password";
+    ConnectionConfiguration config = new ConnectionConfiguration(null, null, null, password);
+    assertThat(config.getPassword()).isEqualTo(password);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ConnectionManagerUnitTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ConnectionManagerUnitTest.java
new file mode 100644
index 0000000..8358d1c
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/ConnectionManagerUnitTest.java
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.ExpectedException;
+
+import org.apache.geode.cache.Operation;
+import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class ConnectionManagerUnitTest {
+  private static final String REGION_NAME = "testRegion";
+  private static final String TABLE_NAME = "testTable";
+  private static final String CONFIG_NAME = "configName";
+  private static final String KEY_COLUMN = "keyColumn";
+
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  private JdbcConnectorService configService;
+  private ConnectionManager manager;
+  private Connection connection;
+  private ConnectionConfiguration connectionConfig;
+  private RegionMapping mapping;
+  private Object key = new Object();
+  private PdxInstance value = mock(PdxInstance.class);
+
+
+  @Before
+  public void setup() throws Exception {
+    configService = mock(JdbcConnectorService.class);
+    manager = spy(new ConnectionManager(configService));
+    connection = mock(Connection.class);
+
+    connectionConfig = getTestConnectionConfig("name", "url", null, null);
+    doReturn(connection).when(manager).getSQLConnection(connectionConfig);
+
+    mapping = mock(RegionMapping.class);
+    when(mapping.getTableName()).thenReturn(TABLE_NAME);
+  }
+
+  @Test
+  public void getsCorrectMapping() {
+    manager.getMappingForRegion(REGION_NAME);
+    verify(configService).getMappingForRegion(REGION_NAME);
+  }
+
+  @Test
+  public void getsCorrectConnectionConfig() {
+    manager.getConnectionConfig(CONFIG_NAME);
+    verify(configService).getConnectionConfig(CONFIG_NAME);
+  }
+
+  @Test
+  public void retrievesANewConnection() throws Exception {
+    Connection returnedConnection = manager.getConnection(connectionConfig);
+    assertThat(returnedConnection).isNotNull().isSameAs(connection);
+  }
+
+  @Test
+  public void retrievesSameConnectionForSameConnectionConfig() throws Exception {
+    Connection returnedConnection = manager.getConnection(connectionConfig);
+    Connection secondReturnedConnection = manager.getConnection(connectionConfig);
+    assertThat(returnedConnection).isNotNull().isSameAs(connection);
+    assertThat(secondReturnedConnection).isNotNull().isSameAs(connection);
+  }
+
+  @Test
+  public void retrievesDifferentConnectionForEachConfig() throws Exception {
+    Connection secondConnection = mock(Connection.class);
+    ConnectionConfiguration secondConnectionConfig =
+        getTestConnectionConfig("newName", "url", null, null);
+    doReturn(secondConnection).when(manager).getSQLConnection(secondConnectionConfig);
+
+    Connection returnedConnection = manager.getConnection(connectionConfig);
+    Connection secondReturnedConnection = manager.getConnection(secondConnectionConfig);
+    assertThat(returnedConnection).isNotNull().isSameAs(connection);
+    assertThat(secondReturnedConnection).isNotNull().isSameAs(secondConnection);
+    assertThat(returnedConnection).isNotSameAs(secondReturnedConnection);
+  }
+
+  @Test
+  public void retrievesANewConnectionIfCachedOneIsClosed() throws Exception {
+    manager.getConnection(connectionConfig);
+    when(connection.isClosed()).thenReturn(true);
+
+    Connection secondConnection = mock(Connection.class);
+    doReturn(secondConnection).when(manager).getSQLConnection(connectionConfig);
+    Connection secondReturnedConnection = manager.getConnection(connectionConfig);
+    assertThat(secondReturnedConnection).isSameAs(secondConnection);
+  }
+
+  @Test
+  public void closesAllConnections() throws Exception {
+    Connection secondConnection = mock(Connection.class);
+    ConnectionConfiguration secondConnectionConfig =
+        getTestConnectionConfig("newName", "url", null, null);
+
+    doReturn(secondConnection).when(manager).getSQLConnection(secondConnectionConfig);
+    manager.getConnection(connectionConfig);
+    manager.getConnection(secondConnectionConfig);
+
+    manager.close();
+    verify(connection).close();
+    verify(secondConnection).close();
+  }
+
+  @Test
+  public void returnsCorrectColumnForDestroy() throws Exception {
+    ResultSet primaryKeys = getPrimaryKeysMetadData();
+    when(primaryKeys.next()).thenReturn(true).thenReturn(false);
+
+    List<ColumnValue> columnValueList =
+        manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.DESTROY);
+    assertThat(columnValueList).hasSize(1);
+    assertThat(columnValueList.get(0).getColumnName()).isEqualTo(KEY_COLUMN);
+  }
+
+  @Test
+  public void returnsCorrectColumnForGet() throws Exception {
+    ResultSet primaryKeys = getPrimaryKeysMetadData();
+    when(primaryKeys.next()).thenReturn(true).thenReturn(false);
+
+    List<ColumnValue> columnValueList =
+        manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.GET);
+    assertThat(columnValueList).hasSize(1);
+    assertThat(columnValueList.get(0).getColumnName()).isEqualTo(KEY_COLUMN);
+  }
+
+  @Test
+  public void throwsExceptionIfTableHasCompositePrimaryKey() throws Exception {
+    ResultSet primaryKeys = getPrimaryKeysMetadData();
+    when(primaryKeys.next()).thenReturn(true);
+
+    thrown.expect(IllegalStateException.class);
+    manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.GET);
+  }
+
+  @Test
+  public void throwsExceptionWhenTwoTablesHasCaseInsensitiveSameName() throws Exception {
+    DatabaseMetaData metadata = mock(DatabaseMetaData.class);
+    when(connection.getMetaData()).thenReturn(metadata);
+    ResultSet resultSet = mock(ResultSet.class);
+    when(resultSet.next()).thenReturn(true).thenReturn(true).thenReturn(false);
+    when(resultSet.getString("TABLE_NAME")).thenReturn(TABLE_NAME);
+    when(resultSet.getString("TABLE_NAME")).thenReturn(TABLE_NAME.toUpperCase());
+    when(metadata.getTables(any(), any(), any(), any())).thenReturn(resultSet);
+
+    thrown.expect(IllegalStateException.class);
+    thrown.expectMessage("Duplicate tables that match region name");
+    manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.GET);
+  }
+
+  @Test
+  public void throwsExceptionWhenDesiredTableNotFound() throws Exception {
+    DatabaseMetaData metadata = mock(DatabaseMetaData.class);
+    when(connection.getMetaData()).thenReturn(metadata);
+    ResultSet resultSet = mock(ResultSet.class);
+    when(resultSet.next()).thenReturn(true).thenReturn(false);
+    when(resultSet.getString("TABLE_NAME")).thenReturn("otherTable");
+    when(metadata.getTables(any(), any(), any(), any())).thenReturn(resultSet);
+
+    thrown.expect(IllegalStateException.class);
+    manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.GET);
+  }
+
+  @Test
+  public void throwsExceptionWhenNoPrimaryKeyInTable() throws Exception {
+    ResultSet primaryKeys = getPrimaryKeysMetadData();
+    when(primaryKeys.next()).thenReturn(false);
+
+    thrown.expect(IllegalStateException.class);
+    manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.GET);
+  }
+
+  @Test
+  public void throwsExceptionWhenFailsToGetTableMetadata() throws Exception {
+    when(connection.getMetaData()).thenThrow(SQLException.class);
+
+    thrown.expect(IllegalStateException.class);
+    manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.GET);
+  }
+
+  @Test
+  public void returnsCorrectColumnsForUpsertOperations() throws Exception {
+    ResultSet primaryKeys = getPrimaryKeysMetadData();
+    when(primaryKeys.next()).thenReturn(true).thenReturn(false);
+
+    String nonKeyColumn = "otherColumn";
+    when(mapping.getColumnNameForField(KEY_COLUMN)).thenReturn(KEY_COLUMN);
+    when(mapping.getColumnNameForField(nonKeyColumn)).thenReturn(nonKeyColumn);
+    when(value.getFieldNames()).thenReturn(Arrays.asList(KEY_COLUMN, nonKeyColumn));
+
+    List<ColumnValue> columnValueList =
+        manager.getColumnToValueList(connectionConfig, mapping, key, value, Operation.UPDATE);
+    assertThat(columnValueList).hasSize(2);
+    assertThat(columnValueList.get(0).getColumnName()).isEqualTo(nonKeyColumn);
+    assertThat(columnValueList.get(1).getColumnName()).isEqualTo(KEY_COLUMN);
+  }
+
+  private ConnectionConfiguration getTestConnectionConfig(String name, String url, String user,
+      String password) {
+    ConnectionConfiguration config = new ConnectionConfiguration(name, url, user, password);
+    return config;
+  }
+
+  private ResultSet getPrimaryKeysMetadData() throws SQLException {
+    DatabaseMetaData metadata = mock(DatabaseMetaData.class);
+    when(connection.getMetaData()).thenReturn(metadata);
+    ResultSet resultSet = mock(ResultSet.class);
+    when(resultSet.next()).thenReturn(true).thenReturn(false);
+    when(resultSet.getString("TABLE_NAME")).thenReturn(TABLE_NAME);
+    when(metadata.getTables(any(), any(), any(), any())).thenReturn(resultSet);
+    ResultSet primaryKeys = mock(ResultSet.class);
+    when(metadata.getPrimaryKeys(any(), any(), anyString())).thenReturn(primaryKeys);
+    when(primaryKeys.getString("COLUMN_NAME")).thenReturn(KEY_COLUMN);
+    return primaryKeys;
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/JdbcConnectorServiceTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/JdbcConnectorServiceTest.java
new file mode 100644
index 0000000..d91e4df
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/JdbcConnectorServiceTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.extension.ExtensionPoint;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class JdbcConnectorServiceTest {
+  private static final String TEST_CONFIG_NAME = "testConfig";
+  private static final String TEST_REGION_NAME = "testRegion";
+
+  private JdbcConnectorService service = new JdbcConnectorService();
+
+  @Before
+  public void setup() {
+    InternalCache cache = mock(InternalCache.class);
+    when(cache.getExtensionPoint()).thenReturn(mock(ExtensionPoint.class));
+    service.init(cache);
+  }
+
+  @Test
+  public void returnsNoConfigIfEmpty() {
+    assertThat(service.getConnectionConfig("foo")).isNull();
+  }
+
+  @Test
+  public void returnsNoMappingIfEmpty() {
+    assertThat(service.getMappingForRegion("foo")).isNull();
+  }
+
+  @Test
+  public void returnsCorrectConfig() {
+    ConnectionConfiguration config = mock(ConnectionConfiguration.class);
+    when(config.getName()).thenReturn(TEST_CONFIG_NAME);
+    service.addOrUpdateConnectionConfig(config);
+
+    assertThat(service.getConnectionConfig(TEST_CONFIG_NAME)).isSameAs(config);
+  }
+
+  @Test
+  public void doesNotReturnConfigWithDifferentName() {
+    ConnectionConfiguration config = mock(ConnectionConfiguration.class);
+    when(config.getName()).thenReturn("theOtherConfig");
+    service.addOrUpdateConnectionConfig(config);
+
+    assertThat(service.getConnectionConfig(TEST_CONFIG_NAME)).isNull();
+  }
+
+  @Test
+  public void returnsCorrectMapping() {
+    RegionMapping mapping = mock(RegionMapping.class);
+    when(mapping.getRegionName()).thenReturn(TEST_REGION_NAME);
+    service.addOrUpdateRegionMapping(mapping);
+
+    assertThat(service.getMappingForRegion(TEST_REGION_NAME)).isSameAs(mapping);
+  }
+
+  @Test
+  public void doesNotReturnMappingForDifferentRegion() {
+    RegionMapping mapping = mock(RegionMapping.class);
+    when(mapping.getRegionName()).thenReturn("theOtherMapping");
+    service.addOrUpdateRegionMapping(mapping);
+
+    assertThat(service.getMappingForRegion(TEST_REGION_NAME)).isNull();
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/PreparedStatementCacheTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/PreparedStatementCacheTest.java
new file mode 100644
index 0000000..ed78310
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/PreparedStatementCacheTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.ExpectedException;
+
+import org.apache.geode.cache.Operation;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class PreparedStatementCacheTest {
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  private PreparedStatementCache cache;
+  private Connection connection;
+  private List<ColumnValue> values = new ArrayList<>();
+
+  @Before
+  public void setup() throws SQLException {
+    cache = new PreparedStatementCache();
+    connection = mock(Connection.class);
+    when(connection.prepareStatement(any())).thenReturn(mock(PreparedStatement.class));
+    values.add(mock(ColumnValue.class));
+  }
+
+  @Test
+  public void returnsSameStatementForIdenticalInputs() throws Exception {
+    cache.getPreparedStatement(connection, values, "table1", Operation.UPDATE, 1);
+    cache.getPreparedStatement(connection, values, "table1", Operation.UPDATE, 1);
+    verify(connection, times(1)).prepareStatement(any());
+  }
+
+  @Test
+  public void returnsDifferentStatementForNonIdenticalInputs() throws Exception {
+    cache.getPreparedStatement(connection, values, "table1", Operation.UPDATE, 1);
+    cache.getPreparedStatement(connection, values, "table2", Operation.UPDATE, 1);
+    verify(connection, times(2)).prepareStatement(any());
+  }
+
+  @Test()
+  public void throwsExceptionIfPreparingStatementFails() throws Exception {
+    when(connection.prepareStatement(any())).thenThrow(SQLException.class);
+    thrown.expect(IllegalStateException.class);
+    cache.getPreparedStatement(connection, values, "table1", Operation.UPDATE, 1);
+  }
+
+  @Test
+  public void throwsExceptionIfInvalidOperationGiven() throws Exception {
+    thrown.expect(IllegalArgumentException.class);
+    cache.getPreparedStatement(connection, values, "table1", Operation.REGION_CLOSE, 1);
+
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/RegionMappingTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/RegionMappingTest.java
new file mode 100644
index 0000000..1c7201c
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/RegionMappingTest.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class RegionMappingTest {
+
+  @Test
+  public void initiatedWithNullValues() {
+    RegionMapping mapping = new RegionMapping(null, null, null, null, false, null);
+    assertThat(mapping.getTableName()).isNull();
+    assertThat(mapping.getRegionName()).isNull();
+    assertThat(mapping.getConnectionConfigName()).isNull();
+    assertThat(mapping.getPdxClassName()).isNull();
+  }
+
+  @Test
+  public void hasCorrectTableName() {
+    String name = "name";
+    RegionMapping mapping = new RegionMapping(null, null, name, null, false, null);
+    assertThat(mapping.getTableName()).isEqualTo(name);
+  }
+
+  @Test
+  public void hasCorrectRegionName() {
+    String name = "name";
+    RegionMapping mapping = new RegionMapping(name, null, null, null, false, null);
+    assertThat(mapping.getRegionName()).isEqualTo(name);
+  }
+
+  @Test
+  public void hasCorrectConfigName() {
+    String name = "name";
+    RegionMapping mapping = new RegionMapping(null, null, null, name, false, null);
+    assertThat(mapping.getConnectionConfigName()).isEqualTo(name);
+  }
+
+  @Test
+  public void hasCorrectPdxClassName() {
+    String name = "name";
+    RegionMapping mapping = new RegionMapping(null, name, null, null, false, null);
+    assertThat(mapping.getPdxClassName()).isEqualTo(name);
+  }
+
+  @Test
+  public void primaryKeyInValueSetCorrectly() {
+    RegionMapping mapping = new RegionMapping(null, null, null, null, true, null);
+    assertThat(mapping.isPrimaryKeyInValue()).isTrue();
+  }
+
+  @Test
+  public void returnsFieldNameIfColumnNotMapped() {
+    String fieldName = "myField";
+    Map<String, String> fieldMap = new HashMap<>();
+    fieldMap.put("otherField", "column");
+    RegionMapping mapping = new RegionMapping(null, null, null, null, true, fieldMap);
+    assertThat(mapping.getColumnNameForField(fieldName)).isEqualTo(fieldName);
+  }
+
+  @Test
+  public void returnsMappedColumnNameForField() {
+    String fieldName = "myField";
+    String columnName = "myColumn";
+    Map<String, String> fieldMap = new HashMap<>();
+    fieldMap.put(fieldName, columnName);
+    RegionMapping mapping = new RegionMapping(null, null, null, null, true, fieldMap);
+    assertThat(mapping.getColumnNameForField(fieldName)).isEqualTo(columnName);
+  }
+
+  @Test
+  public void returnsAllMappings() {
+    String fieldName1 = "myField1";
+    String columnName1 = "myColumn1";
+    String fieldName2 = "myField2";
+    String columnName2 = "myColumn2";
+    Map<String, String> fieldMap = new HashMap<>();
+    fieldMap.put(fieldName1, columnName1);
+    fieldMap.put(fieldName2, columnName2);
+    RegionMapping mapping = new RegionMapping(null, null, null, null, true, fieldMap);
+
+    assertThat(mapping.getFieldToColumnMap().size()).isEqualTo(2);
+    assertThat(mapping.getFieldToColumnMap()).containsOnlyKeys(fieldName1, fieldName2);
+    assertThat(mapping.getFieldToColumnMap()).containsEntry(fieldName1, columnName1)
+        .containsEntry(fieldName2, columnName2);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/SqlHandlerTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/SqlHandlerTest.java
new file mode 100644
index 0000000..7cd623b
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/SqlHandlerTest.java
@@ -0,0 +1,365 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.ExpectedException;
+
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.pdx.PdxInstanceFactory;
+import org.apache.geode.pdx.internal.PdxInstanceImpl;
+import org.apache.geode.pdx.internal.PdxType;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class SqlHandlerTest {
+  private static final String REGION_NAME = "testRegion";
+  private static final String TABLE_NAME = "testTable";
+  private static final Object COLUMN_VALUE_1 = new Object();
+  private static final String COLUMN_NAME_1 = "columnName1";
+  private static final Object COLUMN_VALUE_2 = new Object();
+  private static final String COLUMN_NAME_2 = "columnName2";
+
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  private ConnectionManager manager;
+  private Region region;
+  private InternalCache cache;
+  private SqlHandler handler;
+  private PreparedStatement statement;
+  private RegionMapping regionMapping;
+  private PdxInstanceImpl value;
+
+  @Before
+  public void setup() throws Exception {
+    manager = mock(ConnectionManager.class);
+    region = mock(Region.class);
+    cache = mock(InternalCache.class);
+    when(region.getRegionService()).thenReturn(cache);
+    handler = new SqlHandler(manager);
+    value = mock(PdxInstanceImpl.class);
+    when(value.getPdxType()).thenReturn(mock(PdxType.class));
+    setupManagerMock();
+  }
+
+  @Test
+  public void readReturnsNullIfNoKeyProvided() {
+    thrown.expect(IllegalArgumentException.class);
+    handler.read(region, null);
+  }
+
+  @Test
+  public void usesPdxFactoryForClassWhenExists() throws Exception {
+    setupEmptyResultSet();
+    String pdxClassName = "classname";
+    when(regionMapping.getPdxClassName()).thenReturn(pdxClassName);
+    handler.read(region, new Object());
+
+    verify(cache).createPdxInstanceFactory(pdxClassName);
+    verifyNoMoreInteractions(cache);
+  }
+
+  @Test
+  public void readClearsPreparedStatementWhenFinished() throws Exception {
+    setupEmptyResultSet();
+    handler.read(region, new Object());
+    verify(statement).clearParameters();
+  }
+
+  @Test
+  public void usesPbxFactoryForNoPbxClassWhenClassNonExistent() throws Exception {
+    setupEmptyResultSet();
+    handler.read(region, new Object());
+
+    verify(cache).createPdxInstanceFactory("no class", false);
+    verifyNoMoreInteractions(cache);
+  }
+
+  @Test
+  public void readReturnsNullIfNoResultsReturned() throws Exception {
+    setupEmptyResultSet();
+    assertThat(handler.read(region, new Object())).isNull();
+  }
+
+  @Test
+  public void throwsExceptionIfQueryFails() throws Exception {
+    when(statement.executeQuery()).thenThrow(SQLException.class);
+
+    thrown.expect(IllegalStateException.class);
+    handler.read(region, new Object());
+  }
+
+  @Test
+  public void readReturnsDataFromAllResultColumns() throws Exception {
+    ResultSet result = mock(ResultSet.class);
+    setupResultSet(result);
+    when(result.next()).thenReturn(true).thenReturn(false);
+    when(statement.executeQuery()).thenReturn(result);
+
+    when(manager.getKeyColumnName(any(), anyString())).thenReturn("key");
+    PdxInstanceFactory factory = mock(PdxInstanceFactory.class);
+    when(cache.createPdxInstanceFactory(anyString(), anyBoolean())).thenReturn(factory);
+
+    String filedName1 = COLUMN_NAME_1.toLowerCase();
+    String filedName2 = COLUMN_NAME_2.toLowerCase();
+    handler.read(region, new Object());
+    verify(factory).writeField(filedName1, COLUMN_VALUE_1, Object.class);
+    verify(factory).writeField(filedName2, COLUMN_VALUE_2, Object.class);
+    verify(factory).create();
+  }
+
+  @Test
+  public void readResultOmitsKeyColumnIfNotInValue() throws Exception {
+    ResultSet result = mock(ResultSet.class);
+    setupResultSet(result);
+    when(result.next()).thenReturn(true).thenReturn(false);
+    when(statement.executeQuery()).thenReturn(result);
+
+    when(manager.getKeyColumnName(any(), anyString())).thenReturn(COLUMN_NAME_1);
+    PdxInstanceFactory factory = mock(PdxInstanceFactory.class);
+    when(cache.createPdxInstanceFactory(anyString(), anyBoolean())).thenReturn(factory);
+
+    String filedName2 = COLUMN_NAME_2.toLowerCase();
+    handler.read(region, new Object());
+    verify(factory).writeField(filedName2, COLUMN_VALUE_2, Object.class);
+    verify(factory, times(1)).writeField(any(), any(), any());
+    verify(factory).create();
+  }
+
+  @Test
+  public void throwsExceptionIfMoreThatOneResultReturned() throws Exception {
+    ResultSet result = mock(ResultSet.class);
+    setupResultSet(result);
+    when(result.next()).thenReturn(true);
+    when(result.getStatement()).thenReturn(mock(PreparedStatement.class));
+    when(statement.executeQuery()).thenReturn(result);
+
+    when(manager.getKeyColumnName(any(), anyString())).thenReturn("key");
+    when(cache.createPdxInstanceFactory(anyString(), anyBoolean()))
+        .thenReturn(mock(PdxInstanceFactory.class));
+
+    thrown.expect(IllegalStateException.class);
+    handler.read(region, new Object());
+  }
+
+  @Test
+  public void writeThrowsExceptionIfValueIsNullAndNotDoingDestroy() {
+    thrown.expect(IllegalArgumentException.class);
+    handler.write(region, Operation.UPDATE, new Object(), null);
+  }
+
+  @Test
+  public void insertActionSucceeds() throws Exception {
+    when(statement.executeUpdate()).thenReturn(1);
+    handler.write(region, Operation.CREATE, new Object(), value);
+    verify(statement).setObject(1, COLUMN_VALUE_1);
+    verify(statement).setObject(2, COLUMN_VALUE_2);
+  }
+
+  @Test
+  public void updateActionSucceeds() throws Exception {
+    when(statement.executeUpdate()).thenReturn(1);
+    handler.write(region, Operation.UPDATE, new Object(), value);
+    verify(statement).setObject(1, COLUMN_VALUE_1);
+    verify(statement).setObject(2, COLUMN_VALUE_2);
+  }
+
+  @Test
+  public void destroyActionSucceeds() throws Exception {
+    List<ColumnValue> columnList = new ArrayList<>();
+    columnList.add(new ColumnValue(true, COLUMN_NAME_1, COLUMN_VALUE_1));
+    when(manager.getColumnToValueList(any(), any(), any(), any(), any())).thenReturn(columnList);
+    when(statement.executeUpdate()).thenReturn(1);
+    handler.write(region, Operation.DESTROY, new Object(), value);
+    verify(statement).setObject(1, COLUMN_VALUE_1);
+    verify(statement, times(1)).setObject(anyInt(), any());
+  }
+
+  @Test
+  public void destroyActionThatRemovesNoRowCompletesUnexceptionally() throws Exception {
+    List<ColumnValue> columnList = new ArrayList<>();
+    columnList.add(new ColumnValue(true, COLUMN_NAME_1, COLUMN_VALUE_1));
+    when(manager.getColumnToValueList(any(), any(), any(), any(), any())).thenReturn(columnList);
+    when(statement.executeUpdate()).thenReturn(0);
+    handler.write(region, Operation.DESTROY, new Object(), value);
+    verify(statement).setObject(1, COLUMN_VALUE_1);
+    verify(statement, times(1)).setObject(anyInt(), any());
+  }
+
+  @Test
+  public void destroyThrowExceptionWhenFail() throws Exception {
+    List<ColumnValue> columnList = new ArrayList<>();
+    columnList.add(new ColumnValue(true, COLUMN_NAME_1, COLUMN_VALUE_1));
+    when(manager.getColumnToValueList(any(), any(), any(), any(), any())).thenReturn(columnList);
+    when(statement.executeUpdate()).thenThrow(SQLException.class);
+
+    thrown.expect(IllegalStateException.class);
+    handler.write(region, Operation.DESTROY, new Object(), value);
+  }
+
+  @Test
+  public void preparedStatementClearedAfterExecution() throws Exception {
+    when(statement.executeUpdate()).thenReturn(1);
+    handler.write(region, Operation.CREATE, new Object(), value);
+    verify(statement).clearParameters();
+  }
+
+  @Test
+  public void whenInsertFailsUpdateSucceeds() throws Exception {
+    when(statement.executeUpdate()).thenReturn(0);
+
+    PreparedStatement updateStatement = mock(PreparedStatement.class);
+    when(updateStatement.executeUpdate()).thenReturn(1);
+    when(manager.getPreparedStatement(any(), any(), any(), any(), anyInt())).thenReturn(statement)
+        .thenReturn(updateStatement);
+
+    handler.write(region, Operation.CREATE, new Object(), value);
+    verify(statement).executeUpdate();
+    verify(updateStatement).executeUpdate();
+    verify(statement).clearParameters();
+    verify(updateStatement).clearParameters();
+  }
+
+  @Test
+  public void whenUpdateFailsInsertSucceeds() throws Exception {
+    when(statement.executeUpdate()).thenReturn(0);
+
+    PreparedStatement insertStatement = mock(PreparedStatement.class);
+    when(insertStatement.executeUpdate()).thenReturn(1);
+    when(manager.getPreparedStatement(any(), any(), any(), any(), anyInt())).thenReturn(statement)
+        .thenReturn(insertStatement);
+
+    handler.write(region, Operation.UPDATE, new Object(), value);
+    verify(statement).executeUpdate();
+    verify(insertStatement).executeUpdate();
+    verify(statement).clearParameters();
+    verify(insertStatement).clearParameters();
+  }
+
+  @Test
+  public void whenInsertFailsWithExceptionUpdateSucceeds() throws Exception {
+    when(statement.executeUpdate()).thenThrow(SQLException.class);
+
+    PreparedStatement updateStatement = mock(PreparedStatement.class);
+    when(updateStatement.executeUpdate()).thenReturn(1);
+    when(manager.getPreparedStatement(any(), any(), any(), any(), anyInt())).thenReturn(statement)
+        .thenReturn(updateStatement);
+
+    handler.write(region, Operation.CREATE, new Object(), value);
+    verify(statement).executeUpdate();
+    verify(updateStatement).executeUpdate();
+    verify(statement).clearParameters();
+    verify(updateStatement).clearParameters();
+  }
+
+  @Test
+  public void whenUpdateFailsWithExceptionInsertSucceeds() throws Exception {
+    when(statement.executeUpdate()).thenThrow(SQLException.class);
+
+    PreparedStatement insertStatement = mock(PreparedStatement.class);
+    when(insertStatement.executeUpdate()).thenReturn(1);
+    when(manager.getPreparedStatement(any(), any(), any(), any(), anyInt())).thenReturn(statement)
+        .thenReturn(insertStatement);
+
+    handler.write(region, Operation.UPDATE, new Object(), value);
+    verify(statement).executeUpdate();
+    verify(insertStatement).executeUpdate();
+    verify(statement).clearParameters();
+    verify(insertStatement).clearParameters();
+  }
+
+  @Test
+  public void whenBothInsertAndUpdateFailExceptionIsThrown() throws Exception {
+    when(statement.executeUpdate()).thenThrow(SQLException.class);
+
+    PreparedStatement insertStatement = mock(PreparedStatement.class);
+    when(insertStatement.executeUpdate()).thenThrow(SQLException.class);
+    when(manager.getPreparedStatement(any(), any(), any(), any(), anyInt())).thenReturn(statement)
+        .thenReturn(insertStatement);
+
+    thrown.expect(IllegalStateException.class);
+    handler.write(region, Operation.UPDATE, new Object(), value);
+    verify(statement).clearParameters();
+    verify(insertStatement).clearParameters();
+  }
+
+  @Test
+  public void whenStatementUpdatesMultipleRowsExceptionThrown() throws Exception {
+    when(statement.executeUpdate()).thenReturn(2);
+    thrown.expect(IllegalStateException.class);
+    handler.write(region, Operation.CREATE, new Object(), value);
+    verify(statement).clearParameters();
+  }
+
+  private void setupManagerMock() throws SQLException {
+    ConnectionConfiguration connectionConfig = mock(ConnectionConfiguration.class);
+    when(manager.getConnectionConfig(any())).thenReturn(connectionConfig);
+
+    regionMapping = mock(RegionMapping.class);
+    when(regionMapping.getRegionName()).thenReturn(REGION_NAME);
+    when(regionMapping.getTableName()).thenReturn(TABLE_NAME);
+    when(manager.getMappingForRegion(any())).thenReturn(regionMapping);
+
+    List<ColumnValue> columnList = new ArrayList<>();
+    columnList.add(new ColumnValue(true, COLUMN_NAME_1, COLUMN_VALUE_1));
+    columnList.add(new ColumnValue(true, COLUMN_NAME_2, COLUMN_VALUE_2));
+    when(manager.getColumnToValueList(any(), any(), any(), any(), any())).thenReturn(columnList);
+
+    statement = mock(PreparedStatement.class);
+    when(manager.getPreparedStatement(any(), any(), any(), any(), anyInt())).thenReturn(statement);
+  }
+
+  private void setupResultSet(ResultSet result) throws SQLException {
+    ResultSetMetaData metaData = mock(ResultSetMetaData.class);
+    when(result.getMetaData()).thenReturn(metaData);
+    when(metaData.getColumnCount()).thenReturn(2);
+
+    when(result.getObject(1)).thenReturn(COLUMN_VALUE_1);
+    when(metaData.getColumnName(1)).thenReturn(COLUMN_NAME_1);
+
+    when(result.getObject(2)).thenReturn(COLUMN_VALUE_2);
+    when(metaData.getColumnName(2)).thenReturn(COLUMN_NAME_2);
+  }
+
+  private void setupEmptyResultSet() throws SQLException {
+    ResultSet result = mock(ResultSet.class);
+    when(result.next()).thenReturn(false);
+    when(statement.executeQuery()).thenReturn(result);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/SqlStatementFactoryTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/SqlStatementFactoryTest.java
new file mode 100644
index 0000000..abfaf66
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/SqlStatementFactoryTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class SqlStatementFactoryTest {
+  private static final String TABLE_NAME = "testTable";
+  private static final String KEY_COLUMN_NAME = "column1";
+
+  private List<ColumnValue> columnValues = new ArrayList<>();
+  private SqlStatementFactory factory = new SqlStatementFactory();
+
+  @Before
+  public void setup() {
+    columnValues.add(new ColumnValue(false, "column0", null));
+    columnValues.add(new ColumnValue(true, KEY_COLUMN_NAME, null));
+    columnValues.add(new ColumnValue(false, "column2", null));
+  }
+
+  @Test
+  public void getSelectQueryString() throws Exception {
+    List<ColumnValue> keyColumn = new ArrayList<>();
+    keyColumn.add(new ColumnValue(true, KEY_COLUMN_NAME, null));
+    String statement = factory.createSelectQueryString(TABLE_NAME, keyColumn);
+    String expectedStatement =
+        String.format("SELECT * FROM %s WHERE %s = ?", TABLE_NAME, KEY_COLUMN_NAME);
+    assertThat(statement).isEqualTo(expectedStatement);
+  }
+
+  @Test
+  public void getDestroySqlString() throws Exception {
+    List<ColumnValue> keyColumn = new ArrayList<>();
+    keyColumn.add(new ColumnValue(true, KEY_COLUMN_NAME, null));
+    String statement = factory.createDestroySqlString(TABLE_NAME, keyColumn);
+    String expectedStatement =
+        String.format("DELETE FROM %s WHERE %s = ?", TABLE_NAME, KEY_COLUMN_NAME);
+    assertThat(statement).isEqualTo(expectedStatement);
+  }
+
+  @Test
+  public void getUpdateSqlString() throws Exception {
+    String statement = factory.createUpdateSqlString(TABLE_NAME, columnValues);
+    String expectedStatement = String.format("UPDATE %s SET %s = ?, %s = ? WHERE %s = ?",
+        TABLE_NAME, columnValues.get(0).getColumnName(), columnValues.get(2).getColumnName(),
+        KEY_COLUMN_NAME);
+    assertThat(statement).isEqualTo(expectedStatement);
+  }
+
+  @Test
+  public void getInsertSqlString() throws Exception {
+    String statement = factory.createInsertSqlString(TABLE_NAME, columnValues);
+    String expectedStatement = String.format("INSERT INTO %s (%s, %s, %s) VALUES (?,?,?)",
+        TABLE_NAME, columnValues.get(0).getColumnName(), columnValues.get(1).getColumnName(),
+        columnValues.get(2).getColumnName());
+    assertThat(statement).isEqualTo(expectedStatement);
+  }
+
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/TestConfigService.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/TestConfigService.java
new file mode 100644
index 0000000..b387d72
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/TestConfigService.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.extension.ExtensionPoint;
+
+/**
+ * Generates fake JdbcConnectorService with Connections and RegionMappings for tests.
+ */
+public class TestConfigService {
+  private static final String DB_NAME = "DerbyDB";
+  private static final String REGION_TABLE_NAME = "employees";
+  private static final String REGION_NAME = "employees";
+  private static final String CONNECTION_URL = "jdbc:derby:memory:" + DB_NAME + ";create=true";
+  private static final String CONNECTION_CONFIG_NAME = "testConnectionConfig";
+
+  public static JdbcConnectorService getTestConfigService() {
+    InternalCache cache = mock(InternalCache.class);
+    when(cache.getExtensionPoint()).thenReturn(mock(ExtensionPoint.class));
+
+    JdbcConnectorService service = new JdbcConnectorService();
+    service.init(cache);
+    service.addOrUpdateConnectionConfig(createConnectionConfig());
+    service.addOrUpdateRegionMapping(createRegionMapping());
+    return service;
+  }
+
+  private static RegionMapping createRegionMapping() {
+    return new RegionMapping(REGION_NAME, null, REGION_TABLE_NAME, CONNECTION_CONFIG_NAME, false,
+        Collections.emptyMap());
+  }
+
+  private static ConnectionConfiguration createConnectionConfig() {
+    return new ConnectionConfiguration(CONNECTION_CONFIG_NAME, CONNECTION_URL, null, null);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/TestableConnectionManager.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/TestableConnectionManager.java
new file mode 100644
index 0000000..01f73f8
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/TestableConnectionManager.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal;
+
+public class TestableConnectionManager extends ConnectionManager {
+
+  public TestableConnectionManager(InternalJdbcConnectorService configService) {
+    super(configService);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/ConnectionConfigBuilderTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/ConnectionConfigBuilderTest.java
new file mode 100644
index 0000000..e97ecc1
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/ConnectionConfigBuilderTest.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class ConnectionConfigBuilderTest {
+  @Test
+  public void createsAllNullObjectIfNothingSet() {
+    ConnectionConfiguration config = new ConnectionConfigBuilder().build();
+    assertThat(config.getName()).isNull();
+    assertThat(config.getUrl()).isNull();
+    assertThat(config.getUser()).isNull();
+    assertThat(config.getPassword()).isNull();
+  }
+
+  @Test
+  public void createsObjectWithCorrectValues() {
+    ConnectionConfigBuilder builder = new ConnectionConfigBuilder();
+    builder.withName("name").withUrl("url").withUser("user").withPassword("password");
+    ConnectionConfiguration config = builder.build();
+    assertThat(config.getName()).isEqualTo("name");
+    assertThat(config.getUrl()).isEqualTo("url");
+    assertThat(config.getUser()).isEqualTo("user");
+    assertThat(config.getPassword()).isEqualTo("password");
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/ElementTypeTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/ElementTypeTest.java
new file mode 100644
index 0000000..8c290b7
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/ElementTypeTest.java
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.apache.geode.connectors.jdbc.internal.xml.ElementType.CONNECTION;
+import static org.apache.geode.connectors.jdbc.internal.xml.ElementType.CONNECTION_SERVICE;
+import static org.apache.geode.connectors.jdbc.internal.xml.ElementType.FIELD_MAPPING;
+import static org.apache.geode.connectors.jdbc.internal.xml.ElementType.REGION_MAPPING;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.COLUMN_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.CONNECTION_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.FIELD_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PDX_CLASS;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PRIMARY_KEY_IN_VALUE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.REGION;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.TABLE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.Stack;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.xml.sax.Attributes;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheXmlException;
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.internal.cache.extension.ExtensionPoint;
+import org.apache.geode.internal.cache.xmlcache.CacheCreation;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class ElementTypeTest {
+  private Stack<Object> stack = new Stack<>();
+  private Attributes attributes;
+  private CacheCreation cacheCreation;
+  private ExtensionPoint<Cache> extensionPoint;
+
+  @Before
+  public void setup() {
+    attributes = mock(Attributes.class);
+    cacheCreation = mock(CacheCreation.class);
+    extensionPoint = mock(ExtensionPoint.class);
+    when(cacheCreation.getExtensionPoint()).thenReturn(extensionPoint);
+  }
+
+  @Test
+  public void gettingElementTypeByNameReturnsCorrectType() {
+    assertThat(ElementType.getTypeFromName(CONNECTION_SERVICE.getTypeName()))
+        .isSameAs(CONNECTION_SERVICE);
+    assertThat(ElementType.getTypeFromName(CONNECTION.getTypeName())).isSameAs(CONNECTION);
+    assertThat(ElementType.getTypeFromName(REGION_MAPPING.getTypeName())).isSameAs(REGION_MAPPING);
+    assertThat(ElementType.getTypeFromName(FIELD_MAPPING.getTypeName())).isSameAs(FIELD_MAPPING);
+  }
+
+  @Test
+  public void gettingElementTypeThatDoesNotExistThrowsException() {
+    assertThatThrownBy(() -> ElementType.getTypeFromName("non-existant element"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void startElementConnectionServiceThrowsWithoutCacheCreation() {
+    stack.push(new Object());
+    assertThatThrownBy(() -> CONNECTION_SERVICE.startElement(stack, attributes))
+        .isInstanceOf(CacheXmlException.class);
+  }
+
+  @Test
+  public void startElementConnectionService() {
+    stack.push(cacheCreation);
+    CONNECTION_SERVICE.startElement(stack, attributes);
+    verify(extensionPoint, times(1)).addExtension(any(JdbcServiceConfiguration.class));
+    assertThat(stack.peek()).isInstanceOf(JdbcServiceConfiguration.class);
+  }
+
+  @Test
+  public void endElementConnectionService() {
+    stack.push(new Object());
+    CONNECTION_SERVICE.endElement(stack);
+    assertThat(stack).isEmpty();
+  }
+
+  @Test
+  public void startElementConnectionThrowsWithoutJdbcServiceConfiguration() {
+    stack.push(new Object());
+    assertThatThrownBy(() -> CONNECTION.startElement(stack, attributes))
+        .isInstanceOf(CacheXmlException.class);
+  }
+
+  @Test
+  public void startElementConnection() {
+    JdbcServiceConfiguration serviceConfiguration = mock(JdbcServiceConfiguration.class);
+    stack.push(serviceConfiguration);
+
+    when(attributes.getValue(JdbcConnectorServiceXmlParser.NAME)).thenReturn("connectionName");
+    when(attributes.getValue(JdbcConnectorServiceXmlParser.URL)).thenReturn("url");
+    when(attributes.getValue(JdbcConnectorServiceXmlParser.USER)).thenReturn("username");
+    when(attributes.getValue(JdbcConnectorServiceXmlParser.PASSWORD)).thenReturn("secret");
+
+    CONNECTION.startElement(stack, attributes);
+    ConnectionConfiguration config = ((ConnectionConfigBuilder) stack.pop()).build();
+
+    assertThat(config.getName()).isEqualTo("connectionName");
+    assertThat(config.getUrl()).isEqualTo("url");
+    assertThat(config.getUser()).isEqualTo("username");
+    assertThat(config.getPassword()).isEqualTo("secret");
+  }
+
+  @Test
+  public void endElementConnection() {
+    ConnectionConfigBuilder builder = mock(ConnectionConfigBuilder.class);
+    JdbcServiceConfiguration serviceConfiguration = mock(JdbcServiceConfiguration.class);
+    stack.push(serviceConfiguration);
+    stack.push(builder);
+
+    CONNECTION.endElement(stack);
+
+    assertThat(stack.size()).isEqualTo(1);
+    verify(serviceConfiguration, times(1)).addConnectionConfig(any());
+  }
+
+  @Test
+  public void startElementRegionMappingThrowsWithoutJdbcServiceConfiguration() {
+    stack.push(new Object());
+    assertThatThrownBy(() -> REGION_MAPPING.startElement(stack, attributes))
+        .isInstanceOf(CacheXmlException.class);
+  }
+
+  @Test
+  public void startElementRegionMapping() {
+    JdbcServiceConfiguration serviceConfiguration = mock(JdbcServiceConfiguration.class);
+    stack.push(serviceConfiguration);
+
+    when(attributes.getValue(REGION)).thenReturn("region");
+    when(attributes.getValue(CONNECTION_NAME)).thenReturn("connectionName");
+    when(attributes.getValue(TABLE)).thenReturn("table");
+    when(attributes.getValue(PDX_CLASS)).thenReturn("pdxClass");
+    when(attributes.getValue(PRIMARY_KEY_IN_VALUE)).thenReturn("true");
+
+    ElementType.REGION_MAPPING.startElement(stack, attributes);
+
+    RegionMapping regionMapping = ((RegionMappingBuilder) stack.pop()).build();
+    assertThat(regionMapping.getRegionName()).isEqualTo("region");
+    assertThat(regionMapping.getConnectionConfigName()).isEqualTo("connectionName");
+    assertThat(regionMapping.getTableName()).isEqualTo("table");
+    assertThat(regionMapping.getPdxClassName()).isEqualTo("pdxClass");
+    assertThat(regionMapping.isPrimaryKeyInValue()).isEqualTo(true);
+  }
+
+  @Test
+  public void endElementRegionMapping() {
+    RegionMappingBuilder builder = mock(RegionMappingBuilder.class);
+    JdbcServiceConfiguration serviceConfiguration = mock(JdbcServiceConfiguration.class);
+    stack.push(serviceConfiguration);
+    stack.push(builder);
+
+    ElementType.REGION_MAPPING.endElement(stack);
+
+    assertThat(stack.size()).isEqualTo(1);
+    verify(serviceConfiguration, times(1)).addRegionMapping(any());
+  }
+
+  @Test
+  public void startElementFieldMappingThrowsWithoutRegionMappingBuilder() {
+    stack.push(new Object());
+    assertThatThrownBy(() -> FIELD_MAPPING.startElement(stack, attributes))
+        .isInstanceOf(CacheXmlException.class);
+  }
+
+  @Test
+  public void startElementFieldMapping() {
+    RegionMappingBuilder builder = new RegionMappingBuilder();
+    stack.push(builder);
+    when(attributes.getValue(FIELD_NAME)).thenReturn("fieldName");
+    when(attributes.getValue(COLUMN_NAME)).thenReturn("columnName");
+
+    ElementType.FIELD_MAPPING.startElement(stack, attributes);
+    RegionMapping regionMapping = ((RegionMappingBuilder) stack.pop()).build();
+
+    assertThat(regionMapping.getColumnNameForField("fieldName")).isEqualTo("columnName");
+  }
+
+  @Test
+  public void endElementFieldMapping() {
+    RegionMappingBuilder builder = mock(RegionMappingBuilder.class);
+    JdbcServiceConfiguration serviceConfiguration = mock(JdbcServiceConfiguration.class);
+    stack.push(serviceConfiguration);
+    stack.push(builder);
+
+    ElementType.FIELD_MAPPING.endElement(stack);
+
+    assertThat(stack.size()).isEqualTo(2);
+    verifyZeroInteractions(builder);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGeneratorIntegrationTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGeneratorIntegrationTest.java
new file mode 100644
index 0000000..5b6d260
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGeneratorIntegrationTest.java
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlGenerator.PREFIX;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.COLUMN_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.CONNECTION_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.FIELD_NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.NAME;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.NAMESPACE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PASSWORD;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PDX_CLASS;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.PRIMARY_KEY_IN_VALUE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.REGION;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.TABLE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.URL;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.USER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+import javax.xml.parsers.ParserConfigurationException;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+import org.w3c.dom.NodeList;
+import org.xml.sax.SAXException;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.connectors.jdbc.internal.InternalJdbcConnectorService;
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.xmlcache.CacheXmlGenerator;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+
+@Category(IntegrationTest.class)
+public class JdbcConnectorServiceXmlGeneratorIntegrationTest {
+
+  private InternalCache cache;
+
+  @Rule
+  public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Before
+  public void setup() {
+    cache = (InternalCache) new CacheFactory().create();
+  }
+
+  @After
+  public void tearDown() {
+    cache.close();
+  }
+
+  @Test
+  public void cacheGetServiceReturnsJdbcConnectorService() {
+    assertThat(cache.getService(InternalJdbcConnectorService.class)).isNotNull();
+  }
+
+  @Test
+  public void serviceWithoutInformationDoesNotPersist() throws Exception {
+    cache.getService(InternalJdbcConnectorService.class);
+    generateXml();
+    Document document = getCacheXmlDocument();
+    NodeList elements = getElementsByName(document, ElementType.CONNECTION_SERVICE);
+    assertThat(elements.getLength()).isZero();
+  }
+
+  @Test
+  public void serviceWithConnectionsHasCorrectXml() throws Exception {
+    InternalJdbcConnectorService service = cache.getService(InternalJdbcConnectorService.class);
+    ConnectionConfiguration config = new ConnectionConfigBuilder().withName("name").withUrl("url")
+        .withUser("username").withPassword("secret").build();
+    service.addOrUpdateConnectionConfig(config);
+
+    generateXml();
+
+    Document document = getCacheXmlDocument();
+    NodeList serviceElements = getElementsByName(document, ElementType.CONNECTION_SERVICE);
+    assertThat(serviceElements.getLength()).isEqualTo(1);
+
+    Element serviceElement = (Element) serviceElements.item(0);
+    assertThat(serviceElement.getAttribute("xmlns:" + PREFIX)).isEqualTo(NAMESPACE);
+
+    NodeList connectionElements = getElementsByName(document, ElementType.CONNECTION);
+    assertThat(connectionElements.getLength()).isEqualTo(1);
+
+    Element connectionElement = (Element) connectionElements.item(0);
+    assertThat(connectionElement.getAttribute(NAME)).isEqualTo("name");
+    assertThat(connectionElement.getAttribute(URL)).isEqualTo("url");
+    assertThat(connectionElement.getAttribute(USER)).isEqualTo("username");
+    assertThat(connectionElement.getAttribute(PASSWORD)).isEqualTo("secret");
+  }
+
+  @Test
+  public void generatesXmlContainingRegionMapping() throws Exception {
+    InternalJdbcConnectorService service = cache.getService(InternalJdbcConnectorService.class);
+    RegionMappingBuilder regionMappingBuilder = new RegionMappingBuilder()
+        .withRegionName("regionName").withPdxClassName("pdxClassName").withTableName("tableName")
+        .withConnectionConfigName("connectionConfigName").withPrimaryKeyInValue("true");
+    regionMappingBuilder.withFieldToColumnMapping("fieldName1", "columnMapping1");
+    regionMappingBuilder.withFieldToColumnMapping("fieldName2", "columnMapping2");
+    RegionMapping regionMapping = regionMappingBuilder.build();
+    service.addOrUpdateRegionMapping(regionMapping);
+
+    generateXml();
+
+    Document document = getCacheXmlDocument();
+    NodeList serviceElements = getElementsByName(document, ElementType.CONNECTION_SERVICE);
+    assertThat(serviceElements.getLength()).isEqualTo(1);
+
+    NodeList mappingElements = getElementsByName(document, ElementType.REGION_MAPPING);
+    assertThat(mappingElements.getLength()).isEqualTo(1);
+
+    Element mappingElement = (Element) mappingElements.item(0);
+    assertThat(mappingElement.getAttribute(REGION)).isEqualTo("regionName");
+    assertThat(mappingElement.getAttribute(PDX_CLASS)).isEqualTo("pdxClassName");
+    assertThat(mappingElement.getAttribute(TABLE)).isEqualTo("tableName");
+    assertThat(mappingElement.getAttribute(CONNECTION_NAME)).isEqualTo("connectionConfigName");
+    assertThat(mappingElement.getAttribute(PRIMARY_KEY_IN_VALUE)).isEqualTo("true");
+
+    NodeList fieldMappingElements = getElementsByName(mappingElement, ElementType.FIELD_MAPPING);
+    assertThat(fieldMappingElements.getLength()).isEqualTo(2);
+    validatePresenceOfFieldMapping(fieldMappingElements, "fieldName1", "columnMapping1");
+    validatePresenceOfFieldMapping(fieldMappingElements, "fieldName2", "columnMapping2");
+  }
+
+  private void validatePresenceOfFieldMapping(NodeList elements, String fieldName,
+      String columnName) {
+    for (int i = 0; i < elements.getLength(); i++) {
+      Element fieldMapping = (Element) elements.item(i);
+      if (fieldMapping.getAttribute(FIELD_NAME).equals(fieldName)) {
+        assertThat(fieldMapping.getAttribute(COLUMN_NAME)).isEqualTo(columnName);
+        return;
+      }
+    }
+    fail("Field name '" + fieldName + "' did not match those provided");
+  }
+
+  private NodeList getElementsByName(Document document, ElementType elementType) {
+    String name = getTagName(elementType);
+    return document.getElementsByTagName(name);
+  }
+
+  private NodeList getElementsByName(Element element, ElementType elementType) {
+    String name = getTagName(elementType);
+    return element.getElementsByTagName(name);
+  }
+
+  private String getTagName(ElementType elementType) {
+    return PREFIX + ":" + elementType.getTypeName();
+  }
+
+  private Document getCacheXmlDocument()
+      throws IOException, SAXException, ParserConfigurationException {
+    File cacheXml = new File(temporaryFolder.getRoot(), "cache.xml");
+    DocumentBuilderFactory dbFactory = DocumentBuilderFactory.newInstance();
+    DocumentBuilder dBuilder = dbFactory.newDocumentBuilder();
+    Document document = dBuilder.parse(cacheXml);
+    document.getDocumentElement().normalize();
+    return document;
+  }
+
+  private void generateXml() throws IOException {
+    File cacheXml = new File(temporaryFolder.getRoot(), "cache.xml");
+    PrintWriter printWriter = new PrintWriter(new FileWriter(cacheXml));
+    CacheXmlGenerator.generate(cache, printWriter, true, false, false);
+    printWriter.flush();
+  }
+
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGeneratorTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGeneratorTest.java
new file mode 100644
index 0000000..b91bd91
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlGeneratorTest.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.NAMESPACE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class JdbcConnectorServiceXmlGeneratorTest {
+
+  @Test
+  public void returnsCorrectNamespace() {
+    JdbcConnectorServiceXmlGenerator generator = new JdbcConnectorServiceXmlGenerator(null, null);
+    assertThat(generator.getNamespaceUri()).isEqualTo(NAMESPACE);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlIntegrationTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlIntegrationTest.java
new file mode 100644
index 0000000..ac7fcf5
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlIntegrationTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.connectors.jdbc.internal.InternalJdbcConnectorService;
+import org.apache.geode.connectors.jdbc.internal.JdbcConnectorService;
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.xmlcache.CacheXmlGenerator;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+
+@Category(IntegrationTest.class)
+public class JdbcConnectorServiceXmlIntegrationTest {
+
+  private InternalCache cache;
+  private File cacheXml;
+  private ConnectionConfiguration config1;
+  private ConnectionConfiguration config2;
+  private RegionMapping regionMapping1;
+  private RegionMapping regionMapping2;
+
+  @Rule
+  public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Before
+  public void setup() throws Exception {
+    cache = (InternalCache) new CacheFactory().create();
+    configureService();
+    cacheXml = generateXml();
+    cache.close();
+  }
+
+  @After
+  public void tearDown() {
+    cache.close();
+  }
+
+  @Test
+  public void canRecreateJdbcConnectorServiceFromXml() throws Exception {
+    cache =
+        (InternalCache) new CacheFactory().set(CACHE_XML_FILE, cacheXml.getAbsolutePath()).create();
+
+    JdbcConnectorService service =
+        (JdbcConnectorService) cache.getExtensionPoint().getExtensions().iterator().next();
+    assertThat(service.getConnectionConfig(config1.getName())).isEqualTo(config1);
+    assertThat(service.getConnectionConfig(config2.getName())).isEqualTo(config2);
+    assertThat(service.getMappingForRegion(regionMapping1.getRegionName()))
+        .isEqualTo(regionMapping1);
+    assertThat(service.getMappingForRegion(regionMapping2.getRegionName()))
+        .isEqualTo(regionMapping2);
+  }
+
+  private void configureService() {
+    InternalJdbcConnectorService service = cache.getService(InternalJdbcConnectorService.class);
+    config1 = new ConnectionConfigBuilder().withName("connection1").withUrl("url1")
+        .withUser("username1").withPassword("secret1").build();
+    config2 = new ConnectionConfigBuilder().withName("connection2").withUrl("url2")
+        .withUser("username2").withPassword("secret2").build();
+    service.addOrUpdateConnectionConfig(config1);
+    service.addOrUpdateConnectionConfig(config2);
+
+    RegionMappingBuilder regionMappingBuilder1 = new RegionMappingBuilder()
+        .withRegionName("regionName1").withPdxClassName("pdxClassName1").withTableName("tableName1")
+        .withConnectionConfigName("connection1").withPrimaryKeyInValue("true");
+    regionMappingBuilder1.withFieldToColumnMapping("fieldName1", "columnMapping1");
+    regionMappingBuilder1.withFieldToColumnMapping("fieldName2", "columnMapping2");
+    regionMapping1 = regionMappingBuilder1.build();
+
+    RegionMappingBuilder regionMappingBuilder2 = new RegionMappingBuilder()
+        .withRegionName("regionName2").withPdxClassName("pdxClassName2").withTableName("tableName2")
+        .withConnectionConfigName("connection2").withPrimaryKeyInValue("false");
+    regionMappingBuilder1.withFieldToColumnMapping("fieldName3", "columnMapping3");
+    regionMappingBuilder1.withFieldToColumnMapping("fieldName4", "columnMapping4");
+    regionMapping2 = regionMappingBuilder2.build();
+
+    service.addOrUpdateRegionMapping(regionMapping1);
+    service.addOrUpdateRegionMapping(regionMapping2);
+  }
+
+  private File generateXml() throws IOException {
+    File cacheXml = new File(temporaryFolder.getRoot(), "cache.xml");
+    PrintWriter printWriter = new PrintWriter(new FileWriter(cacheXml));
+    CacheXmlGenerator.generate(cache, printWriter, true, false, false);
+    printWriter.flush();
+    return cacheXml;
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlParserTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlParserTest.java
new file mode 100644
index 0000000..95a4d45
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcConnectorServiceXmlParserTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.apache.geode.connectors.jdbc.internal.xml.ElementType.CONNECTION_SERVICE;
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.NAMESPACE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.Stack;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.xml.sax.Attributes;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.internal.cache.extension.ExtensionPoint;
+import org.apache.geode.internal.cache.xmlcache.CacheCreation;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class JdbcConnectorServiceXmlParserTest {
+
+  private Stack<Object> stack = new Stack<>();
+  private Attributes attributes;
+  private CacheCreation cacheCreation;
+  private ExtensionPoint<Cache> extensionPoint;
+
+  @Before
+  public void setup() {
+    attributes = mock(Attributes.class);
+    cacheCreation = mock(CacheCreation.class);
+    extensionPoint = mock(ExtensionPoint.class);
+    when(cacheCreation.getExtensionPoint()).thenReturn(extensionPoint);
+  }
+
+  @Test
+  public void getNamespaceUriReturnsNamespace() {
+    assertThat(new JdbcConnectorServiceXmlParser().getNamespaceUri()).isEqualTo(NAMESPACE);
+  }
+
+  @Test
+  public void startElementCreatesJdbcServiceConfiguration() throws Exception {
+    JdbcConnectorServiceXmlParser parser = new JdbcConnectorServiceXmlParser();
+    stack.push(cacheCreation);
+    parser.setStack(stack);
+
+    parser.startElement(NAMESPACE, CONNECTION_SERVICE.getTypeName(), null, attributes);
+
+    assertThat(stack.pop()).isInstanceOf(JdbcServiceConfiguration.class);
+  }
+
+  @Test
+  public void startElementWithWrongUriDoesNothing() throws Exception {
+    JdbcConnectorServiceXmlParser parser = new JdbcConnectorServiceXmlParser();
+    stack.push(cacheCreation);
+    parser.setStack(stack);
+
+    parser.startElement("wrongNamespace", CONNECTION_SERVICE.getTypeName(), null, attributes);
+
+    assertThat(stack.pop()).isEqualTo(cacheCreation);
+  }
+
+  @Test
+  public void endElementRemovesJdbcServiceConfiguration() throws Exception {
+    JdbcConnectorServiceXmlParser parser = new JdbcConnectorServiceXmlParser();
+    stack.push(cacheCreation);
+    JdbcServiceConfiguration serviceConfiguration = mock(JdbcServiceConfiguration.class);
+    stack.push(serviceConfiguration);
+    parser.setStack(stack);
+
+    parser.endElement(NAMESPACE, CONNECTION_SERVICE.getTypeName(), null);
+
+    assertThat(stack.pop()).isEqualTo(cacheCreation);
+    verifyZeroInteractions(serviceConfiguration);
+  }
+
+  @Test
+  public void endElementRemovesWithWrongUriDoesNothing() throws Exception {
+    JdbcConnectorServiceXmlParser parser = new JdbcConnectorServiceXmlParser();
+    stack.push(cacheCreation);
+    JdbcServiceConfiguration serviceConfiguration = mock(JdbcServiceConfiguration.class);
+    stack.push(serviceConfiguration);
+    parser.setStack(stack);
+
+    parser.endElement("wrongNamespace", CONNECTION_SERVICE.getTypeName(), null);
+
+    assertThat(stack.pop()).isEqualTo(serviceConfiguration);
+    verifyZeroInteractions(serviceConfiguration);
+  }
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcServiceConfigurationTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcServiceConfigurationTest.java
new file mode 100644
index 0000000..939db7c
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/JdbcServiceConfigurationTest.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.apache.geode.connectors.jdbc.internal.xml.JdbcConnectorServiceXmlParser.NAMESPACE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.connectors.jdbc.internal.ConnectionConfiguration;
+import org.apache.geode.connectors.jdbc.internal.InternalJdbcConnectorService;
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.xmlcache.XmlGenerator;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class JdbcServiceConfigurationTest {
+
+  private JdbcServiceConfiguration configuration;
+
+  private InternalCache cache;
+  private InternalJdbcConnectorService service;
+  private ConnectionConfiguration connection1;
+  private ConnectionConfiguration connection2;
+  private RegionMapping mapping1;
+  private RegionMapping mapping2;
+
+  @Before
+  public void setUp() {
+    connection1 = mock(ConnectionConfiguration.class);
+    connection2 = mock(ConnectionConfiguration.class);
+    mapping1 = mock(RegionMapping.class);
+    mapping2 = mock(RegionMapping.class);
+
+    service = mock(InternalJdbcConnectorService.class);
+    cache = mock(InternalCache.class);
+    when(cache.getService(InternalJdbcConnectorService.class)).thenReturn(service);
+
+    configuration = new JdbcServiceConfiguration();
+  }
+
+  @Test
+  public void getXmlGeneratorReturnsJdbcConnectorServiceXmlGenerator() {
+    XmlGenerator<Cache> generator = configuration.getXmlGenerator();
+
+    assertThat(generator).isInstanceOf(JdbcConnectorServiceXmlGenerator.class);
+  }
+
+  @Test
+  public void getXmlGeneratorReturnsGeneratorWithJdbcConnectorNamespace() {
+    XmlGenerator<Cache> generator = configuration.getXmlGenerator();
+
+    assertThat(generator.getNamespaceUri()).isEqualTo(NAMESPACE);
+  }
+
+  @Test
+  public void getXmlGeneratorReturnsEmptyGeneratorByDefault() {
+    JdbcConnectorServiceXmlGenerator generator =
+        (JdbcConnectorServiceXmlGenerator) configuration.getXmlGenerator();
+
+    assertThat(generator.getConnections()).isEmpty();
+    assertThat(generator.getMappings()).isEmpty();
+  }
+
+  @Test
+  public void getXmlGeneratorWithConnections() {
+    configuration.addConnectionConfig(connection1);
+    configuration.addConnectionConfig(connection2);
+
+    JdbcConnectorServiceXmlGenerator generator =
+        (JdbcConnectorServiceXmlGenerator) configuration.getXmlGenerator();
+
+    assertThat(generator.getConnections()).containsExactly(connection1, connection2);
+  }
+
+  @Test
+  public void getXmlGeneratorWithRegionMappings() {
+    configuration.addRegionMapping(mapping1);
+    configuration.addRegionMapping(mapping2);
+
+    JdbcConnectorServiceXmlGenerator generator =
+        (JdbcConnectorServiceXmlGenerator) configuration.getXmlGenerator();
+
+    assertThat(generator.getMappings()).containsExactly(mapping1, mapping2);
+  }
+
+  @Test
+  public void onCreateWithNoConnectionsOrMappings() {
+    configuration.onCreate(cache, cache);
+    verifyZeroInteractions(service);
+  }
+
+  @Test
+  public void onCreateWithConnections() {
+    configuration.addConnectionConfig(connection1);
+    configuration.addConnectionConfig(connection2);
+
+    configuration.onCreate(cache, cache);
+
+    verify(service, times(1)).addOrUpdateConnectionConfig(connection1);
+    verify(service, times(1)).addOrUpdateConnectionConfig(connection2);
+  }
+
+  @Test
+  public void onCreateWithRegionMappings() {
+    configuration.addRegionMapping(mapping1);
+    configuration.addRegionMapping(mapping2);
+
+    configuration.onCreate(cache, cache);
+
+    verify(service, times(1)).addOrUpdateRegionMapping(mapping1);
+    verify(service, times(1)).addOrUpdateRegionMapping(mapping2);
+  }
+
+}
diff --git a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/RegionMappingBuilderTest.java b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/RegionMappingBuilderTest.java
new file mode 100644
index 0000000..9953d95
--- /dev/null
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/xml/RegionMappingBuilderTest.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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.
+ */
+package org.apache.geode.connectors.jdbc.internal.xml;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.connectors.jdbc.internal.RegionMapping;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class RegionMappingBuilderTest {
+
+  @Test
+  public void createsMappingWithDefaultValuesIfNonAreSet() {
+    RegionMapping regionMapping = new RegionMappingBuilder().build();
+
+    assertThat(regionMapping.getRegionName()).isNull();
+    assertThat(regionMapping.getTableName()).isNull();
+    assertThat(regionMapping.getPdxClassName()).isNull();
+    assertThat(regionMapping.getConnectionConfigName()).isNull();
+    assertThat(regionMapping.isPrimaryKeyInValue()).isFalse();
+  }
+
+  @Test
+  public void createsMappingWithSpecifiedValues() {
+    RegionMappingBuilder builder = new RegionMappingBuilder();
+    RegionMapping regionMapping = builder.withTableName("tableName").withRegionName("regionName")
+        .withPrimaryKeyInValue("true").withPdxClassName("pdxClassName")
+        .withConnectionConfigName("configName").withFieldToColumnMapping("fieldName", "columnName")
+        .build();
+
+    assertThat(regionMapping.getRegionName()).isEqualTo("regionName");
+    assertThat(regionMapping.getTableName()).isEqualTo("tableName");
+    assertThat(regionMapping.getPdxClassName()).isEqualTo("pdxClassName");
+    assertThat(regionMapping.getConnectionConfigName()).isEqualTo("configName");
+    assertThat(regionMapping.isPrimaryKeyInValue()).isTrue();
+    assertThat(regionMapping.getColumnNameForField("fieldName")).isEqualTo("columnName");
+  }
+}
diff --git a/geode-core/src/test/java/org/apache/geode/test/fake/Fakes.java b/geode-core/src/test/java/org/apache/geode/test/fake/Fakes.java
index cc4d3d1..f5b9266 100644
--- a/geode-core/src/test/java/org/apache/geode/test/fake/Fakes.java
+++ b/geode-core/src/test/java/org/apache/geode/test/fake/Fakes.java
@@ -35,6 +35,7 @@ import org.apache.geode.distributed.internal.membership.InternalDistributedMembe
 import org.apache.geode.internal.cache.CachePerfStats;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
 import org.apache.geode.internal.security.SecurityService;
+import org.apache.geode.pdx.PdxInstanceFactory;
 
 /**
  * Factory methods for fake objects for use in test.
@@ -62,6 +63,7 @@ public class Fakes {
     InternalDistributedSystem system = mock(InternalDistributedSystem.class);
     DistributionConfig config = mock(DistributionConfig.class);
     DistributionManager distributionManager = mock(DistributionManager.class);
+    PdxInstanceFactory pdxInstanceFactory = mock(PdxInstanceFactory.class);
     CancelCriterion systemCancelCriterion = mock(CancelCriterion.class);
     DSClock clock = mock(DSClock.class);
     LogWriter logger = mock(LogWriter.class);
@@ -81,6 +83,7 @@ public class Fakes {
     when(cache.getCancelCriterion()).thenReturn(systemCancelCriterion);
     when(cache.getCachePerfStats()).thenReturn(mock(CachePerfStats.class));
     when(cache.getSecurityService()).thenReturn(mock(SecurityService.class));
+    when(cache.createPdxInstanceFactory(any())).thenReturn(pdxInstanceFactory);
 
     when(system.getDistributedMember()).thenReturn(member);
     when(system.getConfig()).thenReturn(config);
@@ -122,7 +125,7 @@ public class Fakes {
     when(region.getCache()).thenReturn(cache);
     when(region.getRegionService()).thenReturn(cache);
     when(region.getName()).thenReturn(name);
-
+    when(region.getFullPath()).thenReturn("/" + name);
     return region;
   }
 
diff --git a/settings.gradle b/settings.gradle
index 13a665c..a84f352 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -30,6 +30,7 @@ include 'geode-lucene'
 include 'geode-old-client-support'
 include 'geode-wan'
 include 'geode-cq'
+include 'geode-connectors'
 include 'geode-benchmarks'
 include 'geode-client-protocol'
 include 'extensions/geode-modules'

-- 
To stop receiving notification emails like this one, please contact
"commits@geode.apache.org" <commits@geode.apache.org>.

Mime
View raw message