You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ds...@apache.org on 2017/11/01 20:47:58 UTC
[geode] branch feature/GEODE-3781 updated: added
JDBCSynchronousWriter and integration test Also added code to force a
PdxInstance and tests to validate what happens if the value can not be
converted to pdx.
This is an automated email from the ASF dual-hosted git repository.
dschneider pushed a commit to branch feature/GEODE-3781
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/feature/GEODE-3781 by this push:
new 44be3fa added JDBCSynchronousWriter and integration test Also added code to force a PdxInstance and tests to validate what happens if the value can not be converted to pdx.
44be3fa is described below
commit 44be3fabc0e97cd78cf0195e34acd340721211bd
Author: Darrel Schneider <ds...@pivotal.io>
AuthorDate: Wed Nov 1 13:46:37 2017 -0700
added JDBCSynchronousWriter and integration test
Also added code to force a PdxInstance and tests to validate what happens
if the value can not be converted to pdx.
---
.../geode/connectors/jdbc/JDBCAsyncWriter.java | 27 +++--
...AsyncWriter.java => JDBCSynchronousWriter.java} | 80 +++++++------
.../jdbc/JDBCAsyncWriterIntegrationTest.java | 52 +++++++++
...a => JDBCSynchronousWriterIntegrationTest.java} | 127 ++++++++++++---------
4 files changed, 184 insertions(+), 102 deletions(-)
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
index 8924d56..0d2f8d6 100644
--- 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
@@ -17,6 +17,9 @@ package org.apache.geode.connectors.jdbc;
import java.util.List;
import java.util.Properties;
+import org.apache.geode.CopyHelper;
+import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.SerializedCacheValue;
import org.apache.geode.cache.asyncqueue.AsyncEvent;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.query.internal.DefaultQuery;
@@ -40,6 +43,20 @@ public class JDBCAsyncWriter implements AsyncEventListener {
@Override
public void close() {}
+ /**
+ * precondition: DefaultQuery.setPdxReadSerialized(true)
+ */
+ @SuppressWarnings("rawtypes")
+ private PdxInstance getPdxInstance(AsyncEvent event) {
+ Object v = event.getDeserializedValue();
+ if (!(v instanceof PdxInstance)) {
+ v = CopyHelper.copy(v);
+ }
+ return (PdxInstance) v;
+ }
+
+
+ @SuppressWarnings("rawtypes")
@Override
public boolean processEvents(List<AsyncEvent> events) {
changeTotalEvents(events.size());
@@ -47,13 +64,9 @@ public class JDBCAsyncWriter implements AsyncEventListener {
DefaultQuery.setPdxReadSerialized(true);
try {
for (AsyncEvent event : events) {
- // TODO: in some cases getDeserializedValue may return non-PdxInstance.
- // In that case need to serialize and deserialize.
try {
- PdxInstance value = (PdxInstance) event.getDeserializedValue();
- logger.info("AsyncEventListener event : " + event);
- this.manager.write(event.getRegion(), event.getOperation(), event.getKey(), value);
-
+ this.manager.write(event.getRegion(), event.getOperation(), event.getKey(),
+ getPdxInstance(event));
changeSuccessfulEvents(1);
} catch (RuntimeException ex) {
// TODO improve the following logging
@@ -70,7 +83,7 @@ public class JDBCAsyncWriter implements AsyncEventListener {
public void init(Properties props) {
JDBCConfiguration config = new JDBCConfiguration(props);
this.manager = new JDBCManager(config);
- };
+ }
private synchronized void changeTotalEvents(long delta) {
this.totalEvents += delta;
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/JDBCSynchronousWriter.java
similarity index 50%
copy from geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JDBCAsyncWriter.java
copy to geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JDBCSynchronousWriter.java
index 8924d56..a37b413 100644
--- a/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JDBCAsyncWriter.java
+++ b/geode-connectors/src/main/java/org/apache/geode/connectors/jdbc/JDBCSynchronousWriter.java
@@ -14,77 +14,81 @@
*/
package org.apache.geode.connectors.jdbc;
-import java.util.List;
import java.util.Properties;
-import org.apache.geode.cache.asyncqueue.AsyncEvent;
-import org.apache.geode.cache.asyncqueue.AsyncEventListener;
+import org.apache.geode.CopyHelper;
+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.JDBCConfiguration;
import org.apache.geode.connectors.jdbc.internal.JDBCManager;
-import org.apache.geode.internal.logging.LogService;
import org.apache.geode.pdx.PdxInstance;
-import org.apache.logging.log4j.Logger;
/*
- * This class provides write behind cache semantics for a JDBC data source using AsyncEventListener.
+ * This class provides synchronous write through to a data source using JDBC.
*
* @since Geode 1.4
*/
-public class JDBCAsyncWriter implements AsyncEventListener {
- static final Logger logger = LogService.getLogger();
- private long totalEvents = 0;
- private long successfulEvents = 0;
+public class JDBCSynchronousWriter<K, V> implements CacheWriter<K, V> {
private JDBCManager manager;
@Override
- public void close() {}
+ public void init(Properties props) {
+ JDBCConfiguration config = new JDBCConfiguration(props);
+ this.manager = new JDBCManager(config);
+ }
@Override
- public boolean processEvents(List<AsyncEvent> events) {
- changeTotalEvents(events.size());
+ public void close() {}
+
+ private PdxInstance getPdxNewValue(EntryEvent<K, V> event) {
// TODO: have a better API that lets you do this
DefaultQuery.setPdxReadSerialized(true);
try {
- for (AsyncEvent event : events) {
- // TODO: in some cases getDeserializedValue may return non-PdxInstance.
- // In that case need to serialize and deserialize.
- try {
- PdxInstance value = (PdxInstance) event.getDeserializedValue();
- logger.info("AsyncEventListener event : " + event);
- this.manager.write(event.getRegion(), event.getOperation(), event.getKey(), value);
-
- changeSuccessfulEvents(1);
- } catch (RuntimeException ex) {
- // TODO improve the following logging
- logger.error("Exception processing event " + event, ex);
+ Object v = event.getNewValue();
+ if (!(v instanceof PdxInstance)) {
+ SerializedCacheValue<V> sv = event.getSerializedNewValue();
+ if (sv != null) {
+ v = sv.getDeserializedValue();
+ } else {
+ v = CopyHelper.copy(v);
}
}
+ return (PdxInstance) v;
} finally {
DefaultQuery.setPdxReadSerialized(false);
}
- return true;
}
@Override
- public void init(Properties props) {
- JDBCConfiguration config = new JDBCConfiguration(props);
- this.manager = new JDBCManager(config);
- };
+ public void beforeUpdate(EntryEvent<K, V> event) throws CacheWriterException {
+ this.manager.write(event.getRegion(), event.getOperation(), event.getKey(),
+ getPdxNewValue(event));
+ }
- private synchronized void changeTotalEvents(long delta) {
- this.totalEvents += delta;
+ @Override
+ public void beforeCreate(EntryEvent<K, V> event) throws CacheWriterException {
+ this.manager.write(event.getRegion(), event.getOperation(), event.getKey(),
+ getPdxNewValue(event));
}
- public synchronized long getTotalEvents() {
- return this.totalEvents;
+ @Override
+ public void beforeDestroy(EntryEvent<K, V> event) throws CacheWriterException {
+ this.manager.write(event.getRegion(), event.getOperation(), event.getKey(),
+ getPdxNewValue(event));
}
- private synchronized void changeSuccessfulEvents(long delta) {
- this.successfulEvents += delta;
+ @Override
+ public void beforeRegionDestroy(RegionEvent<K, V> event) throws CacheWriterException {
+ // this event is not sent to JDBC
}
- public synchronized long getSuccessfulEvents() {
- return this.successfulEvents;
+ @Override
+ public void beforeRegionClear(RegionEvent<K, V> event) throws CacheWriterException {
+ // this event is not sent to JDBC
}
+
}
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
index a3674c4..6f5e831 100644
--- 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
@@ -31,7 +31,10 @@ import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.connectors.jdbc.JDBCAsyncWriter;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.pdx.PdxReader;
+import org.apache.geode.pdx.PdxSerializable;
import org.apache.geode.pdx.PdxSerializationException;
+import org.apache.geode.pdx.PdxWriter;
import org.apache.geode.test.junit.categories.IntegrationTest;
import org.awaitility.Awaitility;
@@ -162,6 +165,55 @@ public class JDBCAsyncWriterIntegrationTest {
}
@Test
+ public void putNonPdxInstanceFails() {
+ Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ employees.put("1", "non pdx instance");
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> assertThat(jdbcWriter.getTotalEvents()).isEqualTo(1));
+ assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(0);
+ }
+
+ @Test
+ public void putNonPdxInstanceThatIsPdxSerializable() throws SQLException {
+ Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ Object value = new TestEmployee("Emp2", 22);
+ employees.put("2", value);
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
+
+ ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
+ assertThat(rs.next()).isTrue();
+ assertThat(rs.getString("id")).isEqualTo("2");
+ assertThat(rs.getString("name")).isEqualTo("Emp2");
+ assertThat(rs.getObject("age")).isEqualTo(22);
+ assertThat(rs.next()).isFalse();
+ }
+
+ public static class TestEmployee implements PdxSerializable {
+ private String name;
+ private int age;
+
+ public TestEmployee(String name, int age) {
+ this.name = name;
+ this.age = 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");
+ }
+ }
+
+ @Test
public void canDestroyFromTable() throws Exception {
Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
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/JDBCSynchronousWriterIntegrationTest.java
similarity index 75%
copy from geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JDBCAsyncWriterIntegrationTest.java
copy to geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JDBCSynchronousWriterIntegrationTest.java
index a3674c4..95df9c7 100644
--- a/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JDBCAsyncWriterIntegrationTest.java
+++ b/geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/JDBCSynchronousWriterIntegrationTest.java
@@ -14,12 +14,13 @@
*/
package org.apache.geode.connectors.jdbc;
+import static com.googlecode.catchexception.CatchException.catchException;
+import static com.googlecode.catchexception.CatchException.caughtException;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.assertj.core.api.Assertions.*;
import java.sql.*;
import java.util.Properties;
-import java.util.concurrent.TimeUnit;
import org.junit.*;
import org.junit.experimental.categories.Category;
@@ -28,15 +29,16 @@ 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.JDBCAsyncWriter;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.pdx.PdxReader;
+import org.apache.geode.pdx.PdxSerializable;
import org.apache.geode.pdx.PdxSerializationException;
+import org.apache.geode.pdx.PdxWriter;
import org.apache.geode.test.junit.categories.IntegrationTest;
-import org.awaitility.Awaitility;
@Category(IntegrationTest.class)
-public class JDBCAsyncWriterIntegrationTest {
+public class JDBCSynchronousWriterIntegrationTest {
private Cache cache;
@@ -44,7 +46,8 @@ public class JDBCAsyncWriterIntegrationTest {
private Statement stmt;
- JDBCAsyncWriter jdbcWriter;
+ @SuppressWarnings("rawtypes")
+ JDBCSynchronousWriter jdbcWriter;
private String dbName = "DerbyDB";
@@ -105,22 +108,9 @@ public class JDBCAsyncWriterIntegrationTest {
}
@Test
- public void validateJDBCAsyncWriterTotalEvents() {
- Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
- PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
- .writeInt("age", 55).create();
- PdxInstance pdx2 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp2")
- .writeInt("age", 21).create();
- employees.put("1", pdx1);
- employees.put("2", pdx2);
-
- Awaitility.await().atMost(5, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getTotalEvents()).isEqualTo(2));
- }
-
- @Test
public void canInsertIntoTable() throws Exception {
- Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
.writeInt("age", 55).create();
PdxInstance pdx2 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp2")
@@ -128,9 +118,6 @@ public class JDBCAsyncWriterIntegrationTest {
employees.put("1", pdx1);
employees.put("2", pdx2);
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
-
ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
assertThat(rs.next()).isTrue();
assertThat(rs.getString("id")).isEqualTo("1");
@@ -145,14 +132,12 @@ public class JDBCAsyncWriterIntegrationTest {
@Test
public void verifyThatPdxFieldNamedSameAsPrimaryKeyIsIgnored() throws Exception {
- Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
.writeInt("age", 55).writeInt("id", 3).create();
employees.put("1", pdx1);
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
-
ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
assertThat(rs.next()).isTrue();
assertThat(rs.getString("id")).isEqualTo("1");
@@ -162,8 +147,56 @@ public class JDBCAsyncWriterIntegrationTest {
}
@Test
+ public void putNonPdxInstanceFails() {
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
+ catchException(employees).put("1", "non pdx instance");
+ assertThat((Exception) caughtException()).isInstanceOf(ClassCastException.class);
+ assertThat(caughtException().getMessage())
+ .isEqualTo("java.lang.String cannot be cast to org.apache.geode.pdx.PdxInstance");
+ }
+
+ @Test
+ public void putNonPdxInstanceThatIsPdxSerializable() throws SQLException {
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
+ Object value = new TestEmployee("Emp2", 22);
+ employees.put("2", value);
+
+ ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
+ assertThat(rs.next()).isTrue();
+ assertThat(rs.getString("id")).isEqualTo("2");
+ assertThat(rs.getString("name")).isEqualTo("Emp2");
+ assertThat(rs.getObject("age")).isEqualTo(22);
+ assertThat(rs.next()).isFalse();
+ }
+
+ public static class TestEmployee implements PdxSerializable {
+ private String name;
+ private int age;
+
+ public TestEmployee(String name, int age) {
+ this.name = name;
+ this.age = 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");
+ }
+ }
+
+ @Test
public void canDestroyFromTable() throws Exception {
- Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
.writeInt("age", 55).create();
PdxInstance pdx2 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp2")
@@ -171,9 +204,6 @@ public class JDBCAsyncWriterIntegrationTest {
employees.put("1", pdx1);
employees.put("2", pdx2);
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
-
try {
employees.destroy("1");
} catch (PdxSerializationException ignore) {
@@ -182,9 +212,6 @@ public class JDBCAsyncWriterIntegrationTest {
// does not have a real class
}
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(3));
-
ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
assertThat(rs.next()).isTrue();
assertThat(rs.getString("id")).isEqualTo("2");
@@ -195,14 +222,12 @@ public class JDBCAsyncWriterIntegrationTest {
@Test
public void canUpdateTable() throws Exception {
- Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
.writeInt("age", 55).create();
employees.put("1", pdx1);
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
-
PdxInstance pdx3 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
.writeInt("age", 72).create();
try {
@@ -213,9 +238,6 @@ public class JDBCAsyncWriterIntegrationTest {
// does not have a real class
}
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
-
ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
assertThat(rs.next()).isTrue();
assertThat(rs.getObject("age")).isEqualTo(72);
@@ -224,14 +246,12 @@ public class JDBCAsyncWriterIntegrationTest {
@Test
public void canUpdateBecomeInsert() throws Exception {
- Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
.writeInt("age", 55).create();
employees.put("1", pdx1);
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
-
stmt.execute("delete from " + regionTableName + " where id = '1'");
validateTableRowCount(0);
@@ -245,9 +265,6 @@ public class JDBCAsyncWriterIntegrationTest {
// does not have a real class
}
- Awaitility.await().atMost(10, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(2));
-
ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
assertThat(rs.next()).isTrue();
assertThat(rs.getString("id")).isEqualTo("1");
@@ -261,14 +278,12 @@ public class JDBCAsyncWriterIntegrationTest {
stmt.execute("Insert into " + regionTableName + " values('1', 'bogus', 11)");
validateTableRowCount(1);
- Region employees = createRegionWithJDBCAsyncWriter(regionTableName, getRequiredProperties());
+ Region employees =
+ createRegionWithJDBCSynchronousWriter(regionTableName, getRequiredProperties());
PdxInstance pdx1 = cache.createPdxInstanceFactory("Employee").writeString("name", "Emp1")
.writeInt("age", 55).create();
employees.put("1", pdx1);
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
- .until(() -> assertThat(jdbcWriter.getSuccessfulEvents()).isEqualTo(1));
-
ResultSet rs = stmt.executeQuery("select * from " + regionTableName + " order by id asc");
assertThat(rs.next()).isTrue();
assertThat(rs.getString("id")).isEqualTo("1");
@@ -277,14 +292,12 @@ public class JDBCAsyncWriterIntegrationTest {
assertThat(rs.next()).isFalse();
}
- private Region createRegionWithJDBCAsyncWriter(String regionName, Properties props) {
- jdbcWriter = new JDBCAsyncWriter();
+ private Region createRegionWithJDBCSynchronousWriter(String regionName, Properties props) {
+ jdbcWriter = new JDBCSynchronousWriter();
jdbcWriter.init(props);
- cache.createAsyncEventQueueFactory().setBatchSize(1).setBatchTimeInterval(1)
- .create("jdbcAsyncQueue", jdbcWriter);
RegionFactory rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
- rf.addAsyncEventQueueId("jdbcAsyncQueue");
+ rf.setCacheWriter(jdbcWriter);
return rf.create(regionName);
}
--
To stop receiving notification emails like this one, please contact
['"commits@geode.apache.org" <co...@geode.apache.org>'].