You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2016/05/18 10:46:37 UTC
ignite git commit: Fixed IgniteSourceConnectorTest,
IgniteTxExceptionAbstractSelfTest.
Repository: ignite
Updated Branches:
refs/heads/master fa381389b -> da49670f3
Fixed IgniteSourceConnectorTest, IgniteTxExceptionAbstractSelfTest.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/da49670f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/da49670f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/da49670f
Branch: refs/heads/master
Commit: da49670f3a47b057782262d92af7cdea10323cd1
Parents: fa38138
Author: sboikov <sb...@gridgain.com>
Authored: Wed May 18 13:46:27 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed May 18 13:46:27 2016 +0300
----------------------------------------------------------------------
.../IgniteTxExceptionAbstractSelfTest.java | 37 ++++++-------
.../connect/IgniteSourceConnectorTest.java | 58 ++++++++++++++------
2 files changed, 58 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/da49670f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
index 4826c68..95f1740 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
@@ -54,9 +54,6 @@ import static org.apache.ignite.cache.CacheMode.REPLICATED;
* Tests that transaction is invalidated in case of {@link IgniteTxHeuristicCheckedException}.
*/
public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstractSelfTest {
- /** Index SPI throwing exception. */
- private static TestIndexingSpi idxSpi = new TestIndexingSpi();
-
/** */
private static final int PRIMARY = 0;
@@ -78,7 +75,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
- cfg.setIndexingSpi(idxSpi);
+ cfg.setIndexingSpi(new TestIndexingSpi());
cfg.getTransactionConfiguration().setTxSerializableEnabled(true);
@@ -108,7 +105,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
/** {@inheritDoc} */
@Override protected void afterTest() throws Exception {
- idxSpi.forceFail(false);
+ TestIndexingSpi.forceFail(false);
Transaction tx = jcache().unwrap(Ignite.class).transactions().tx();
@@ -321,7 +318,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
IgniteCache<Integer, Integer> cache = grid(0).cache(null);
if (putBefore) {
- idxSpi.forceFail(false);
+ TestIndexingSpi.forceFail(false);
info("Start transaction.");
@@ -344,7 +341,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
grid(i).cache(null).get(key);
}
- idxSpi.forceFail(true);
+ TestIndexingSpi.forceFail(true);
try {
info("Start transaction.");
@@ -380,7 +377,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
*/
@SuppressWarnings("unchecked")
private void checkUnlocked(final Integer key) throws Exception {
- idxSpi.forceFail(false);
+ TestIndexingSpi.forceFail(false);
awaitPartitionMapExchange();
@@ -448,7 +445,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
*/
private void checkPut(boolean putBefore, final Integer key) throws Exception {
if (putBefore) {
- idxSpi.forceFail(false);
+ TestIndexingSpi.forceFail(false);
info("Put key: " + key);
@@ -459,7 +456,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
for (int i = 0; i < gridCount(); i++)
grid(i).cache(null).get(key);
- idxSpi.forceFail(true);
+ TestIndexingSpi.forceFail(true);
info("Going to put: " + key);
@@ -481,7 +478,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
*/
private void checkTransform(boolean putBefore, final Integer key) throws Exception {
if (putBefore) {
- idxSpi.forceFail(false);
+ TestIndexingSpi.forceFail(false);
info("Put key: " + key);
@@ -492,7 +489,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
for (int i = 0; i < gridCount(); i++)
grid(i).cache(null).get(key);
- idxSpi.forceFail(true);
+ TestIndexingSpi.forceFail(true);
info("Going to transform: " + key);
@@ -524,7 +521,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
assert keys.length > 1;
if (putBefore) {
- idxSpi.forceFail(false);
+ TestIndexingSpi.forceFail(false);
Map<Integer, Integer> m = new HashMap<>();
@@ -542,7 +539,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
grid(i).cache(null).get(key);
}
- idxSpi.forceFail(true);
+ TestIndexingSpi.forceFail(true);
final Map<Integer, Integer> m = new HashMap<>();
@@ -570,7 +567,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
*/
private void checkRemove(boolean putBefore, final Integer key) throws Exception {
if (putBefore) {
- idxSpi.forceFail(false);
+ TestIndexingSpi.forceFail(false);
info("Put key: " + key);
@@ -581,7 +578,7 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
for (int i = 0; i < gridCount(); i++)
grid(i).cache(null).get(key);
- idxSpi.forceFail(true);
+ TestIndexingSpi.forceFail(true);
info("Going to remove: " + key);
@@ -657,13 +654,13 @@ public abstract class IgniteTxExceptionAbstractSelfTest extends GridCacheAbstrac
*/
private static class TestIndexingSpi extends IgniteSpiAdapter implements IndexingSpi {
/** Fail flag. */
- private volatile boolean fail;
+ private static volatile boolean fail;
/**
- * @param fail Fail flag.
+ * @param failFlag Fail flag.
*/
- public void forceFail(boolean fail) {
- this.fail = fail;
+ public static void forceFail(boolean failFlag) {
+ fail = failFlag;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/da49670f/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
index 87b11b9..7cdb09c 100644
--- a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
@@ -25,13 +25,18 @@ import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.stream.kafka.TestKafkaBroker;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -203,7 +208,11 @@ public class IgniteSourceConnectorTest extends GridCommonAbstractTest {
assertEquals(EVENT_CNT, cache.size(CachePeekMode.PRIMARY));
// Checks the events are transferred to Kafka broker.
- checkDataDelivered(conditioned);
+ if (conditioned)
+ checkDataDelivered(EVENT_CNT * TOPICS.length / 2);
+ else
+ checkDataDelivered(EVENT_CNT * TOPICS.length);
+
}
/**
@@ -235,9 +244,10 @@ public class IgniteSourceConnectorTest extends GridCommonAbstractTest {
/**
* Checks if events were delivered to Kafka server.
*
- * @param conditioned Flag indicating whether filtering is enabled.
+ * @param expectedEventsCnt Expected events count.
+ * @throws Exception If failed.
*/
- private void checkDataDelivered(boolean conditioned) {
+ private void checkDataDelivered(final int expectedEventsCnt) throws Exception {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker.getBrokerAddress());
@@ -250,23 +260,40 @@ public class IgniteSourceConnectorTest extends GridCommonAbstractTest {
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.ignite.stream.kafka.connect.serialization.CacheEventDeserializer");
- KafkaConsumer<String, CacheEvent> consumer = new KafkaConsumer<>(props);
+ final KafkaConsumer<String, CacheEvent> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(TOPICS));
- int evtCnt = 0;
- long start = System.currentTimeMillis();
+ final AtomicInteger evtCnt = new AtomicInteger();
try {
- while ((System.currentTimeMillis() - start) < 10000) {
- ConsumerRecords<String, CacheEvent> records = consumer.poll(10);
- for (ConsumerRecord<String, CacheEvent> record : records) {
- System.out.println("Event: offset = " + record.offset() + ", key = " + record.key()
- + ", value = " + record.value().toString());
+ // Wait for expected events count.
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ ConsumerRecords<String, CacheEvent> records = consumer.poll(10);
+ for (ConsumerRecord<String, CacheEvent> record : records) {
+ info("Record: " + record);
+
+ evtCnt.getAndIncrement();
+ }
+ return evtCnt.get() >= expectedEventsCnt;
+ }
+ }, 20_000);
- evtCnt++;
+
+ info("Waiting for unexpected records for 5 secs.");
+
+ assertFalse(GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ ConsumerRecords<String, CacheEvent> records = consumer.poll(10);
+ for (ConsumerRecord<String, CacheEvent> record : records) {
+ error("Unexpected record: " + record);
+
+ evtCnt.getAndIncrement();
+ }
+ return evtCnt.get() > expectedEventsCnt;
}
- }
+ }, 5_000));
}
catch (WakeupException e) {
// ignore for shutdown.
@@ -274,10 +301,7 @@ public class IgniteSourceConnectorTest extends GridCommonAbstractTest {
finally {
consumer.close();
- if (conditioned)
- assertEquals((EVENT_CNT * TOPICS.length) / 2, evtCnt);
- else
- assertEquals(EVENT_CNT * TOPICS.length, evtCnt);
+ assertEquals(expectedEventsCnt, evtCnt.get());
}
}