import org.apache.ignite.IgniteException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public void invalidate(Iterable<Metadata> subPath, final IgniteBiInClosure<String, Set<Object>> action) {
    Map<String, List<Metadata>> names = getSnapshotsByCache(subPath);
    if (!names.isEmpty()) {
        ignite.compute().execute(new ComputeTaskSplitAdapter<Map<String, List<Metadata>>, Void>() {
            /** {@inheritDoc} */
            @Override protected Collection<? extends ComputeJob> split(int gridSize,
                Map<String, List<Metadata>> byCache) throws IgniteException {
                List<ComputeJob> result = new ArrayList<>();
                for (Map.Entry<String, List<Metadata>> entry : byCache.entrySet()) {
                    String cacheName = entry.getKey();
                    for (Metadata metadata : entry.getValue()) {
                        result.add(new ProcessAllKeysJob(cacheName, metadata, action));
                    }
                }
                return result;
            }

            /** {@inheritDoc} */
            @Override public Void reduce(List<ComputeJobResult> results) throws IgniteException {
                return null;
            }
        }, names);
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * Performs restore operaton on given resource names.
 *
 * @param source base path to existing backup.
 * @param names of resources included in this backup.
 */
private void restoreData(final URI source, Iterable<String> names) {
    failOnExistingTransactions();
    ignite.compute().execute(new ComputeTaskSplitAdapter<Iterable<String>, Object>() {
        /** {@inheritDoc} */
        @Override protected Collection<? extends ComputeJob> split(int gridSize,
            Iterable<String> arg) throws IgniteException {
            List<ComputeJob> result = new ArrayList<>();
            for (String name : arg) {
                result.add(new RestoreJob(source, name));
            }
            return result;
        }

        /** {@inheritDoc} */
        @Nullable @Override public Object reduce(List<ComputeJobResult> results) throws IgniteException {
            return null;
        }
    }, names);
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Object execute() throws IgniteException {
    Injection.inject(this, ignite);
    try (Exporter.ReaderProvider provider = exporter.read(source)) {
        try (Exporter.Reader reader = provider.open(name)) {
            if (isData(name)) {
                Cache.Entry<String, Metadata> entry = cacheAndMetadataFor(name);
                if (entry != null) {
                    reader.read(keyValueManager.getDataWriter(entry.getKey(), entry.getValue()));
                }
            }
            return null;
        }
    }
    catch (IOException e) {
        throw new IgniteException(e);
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
private void dropTable(String keyspace, String table) {
    String fullName = keyspace + "." + table;
    int attempt = 0;
    Throwable error = null;
    String errorMsg = "Failed to drop Cassandra table '" + fullName + "'";
    RandomSleeper sleeper = newSleeper();
    String statement = "drop table if exists " + fullName;
    while (attempt < CQL_EXECUTION_ATTEMPTS_COUNT) {
        try {
            session().execute(statement);
            return;
        }
        catch (Throwable e) {
            if (CassandraHelper.isHostsAvailabilityError(e)) {
                handleHostsAvailabilityError(e, attempt, errorMsg);
            }
            else {
                throw new IgniteException(errorMsg, e);
            }
            error = e;
        }
        sleeper.sleep();
        attempt++;
    }
    throw new IgniteException(errorMsg, error);
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
@Override public void truncate(String keyspace, String table) {
    String fullName = keyspace + "." + table;
    int attempt = 0;
    Throwable error = null;
    String errorMsg = "Failed to drop Cassandra table '" + fullName + "'";
    RandomSleeper sleeper = newSleeper();
    String statement = "truncate " + fullName;
    while (attempt < CQL_EXECUTION_ATTEMPTS_COUNT) {
        try {
            session().execute(statement);
            return;
        }
        catch (Throwable e) {
            if (CassandraHelper.isHostsAvailabilityError(e)) {
                handleHostsAvailabilityError(e, attempt, errorMsg);
            }
            else {
                throw new IgniteException(errorMsg, e);
            }
            error = e;
        }
        sleeper.sleep();
        attempt++;
    }
    throw new IgniteException(errorMsg, error);
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * @return Cassandra driver session.
 */
private synchronized Session session() {
    if (ses != null) {
        return ses;
    }
    ses = SessionPool.get(this);
    if (ses != null) {
        return ses;
    }
    synchronized (sesStatements) {
        sesStatements.clear();
    }
    try {
        return ses = builder.build().connect();
    }
    catch (Throwable e) {
        throw new IgniteException("Failed to establish session with Cassandra database", e);
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * Constructs Ignite cache key/value persistence settings.
 *
 * @param settingsRsrc resource containing xml with persistence settings for Ignite cache key/value
 */
public PublicKeyspacePersistenceSettings(Resource settingsRsrc) {
    InputStream in;
    try {
        in = settingsRsrc.getInputStream();
    }
    catch (IOException e) {
        throw new IgniteException("Failed to get input stream for Cassandra persistence settings resource: " +
            settingsRsrc, e);
    }
    try {
        init(loadSettings(in));
    }
    finally {
        U.closeQuiet(in);
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * Constructs Ignite cache key/value persistence settings.
 *
 * @param settingsRsrc resource containing xml with persistence settings for Ignite cache key/value
 */
public PublicKeyValuePersistenceSettings(Resource settingsRsrc) {
    InputStream in;
    try {
        in = settingsRsrc.getInputStream();
    }
    catch (IOException e) {
        throw new IgniteException("Failed to get input stream for Cassandra persistence settings resource: " +
            settingsRsrc, e);
    }
    try {
        init(loadSettings(in));
    }
    finally {
        U.closeQuiet(in);
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Object execute() throws IgniteException {
    Injection.inject(this, ignite);
    final Map<Object, Cache.Entry<?, ?>> mergedData = new HashMap<>();
    for (Metadata metadata : metadatas) {
        keyValueProvider.fetchAllKeyValues(cacheName, metadata, new IgniteInClosure<Cache.Entry<Object, Object>>() {
            @Override public void apply(Cache.Entry<Object, Object> entry) {
                Object key = entry.getKey();
                if (!mergedData.containsKey(key)) {
                    mergedData.put(key, new CacheEntryImpl<>(key, entry.getValue()));
                }
            }
        });
    }
    keyValueProvider.write(idSequencer.getNextId(), Collections.singletonMap(cacheName, mergedData.values()), destination);
    return null;
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/** */
@Test(expected = IgniteException.class)
public void failIfChecksumDoesNotMatch() throws IOException {
    URI exportDirUri = folder.newFolder().toURI();
    String destination = exportDirUri.toString();
    backupInCompute(destination);

    Files.walkFileTree(Paths.get(exportDirUri), new SimpleFileVisitor<Path>() {
        @Override public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
            if (FileChecksumHelper.isChecksum(file)) {
                Files.write(file, "corrupted".getBytes());
            }
            return FileVisitResult.CONTINUE;
        }
    });
    restoreInCompute(destination);
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
@Override
public void stop() throws IgniteException {
    try {
        LOGGER.info("[C] attempt to shutdown executor");
        executor.shutdown();
        executor.awaitTermination(TERMINATION_TIMEOUT, TimeUnit.SECONDS);
    }
    catch (InterruptedException e) {
        LOGGER.warn("[C] tasks interrupted");
    }
    finally {
        if (!executor.isTerminated()) {
            LOGGER.warn("[C] cancel non-finished tasks");
        }
        executor.shutdownNow();
        LOGGER.info("[C] shutdown finished");
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
private void failNotifyTransactionReadAfterRegisteringTxs() {
    drGrid().compute().broadcast(new ActiveStoreIgniteRunnable() {
        @Inject
        private transient DoubleLeadService lead;

        @Override protected void runInjected() {
            lead.setWorkInstead(new LeadServiceAdapter() {
                private boolean transactionsSent = false;

                @Override public LeadResponse notifyTransactionsRead(UUID consumerId,
                    List<TransactionMetadata> metadatas) {
                    if (transactionsSent) {
                        throw new IgniteException();
                    }
                    if (!metadatas.isEmpty()) {
                        transactionsSent = true;
                    }
                    return null;
                }
            });
        }
    });
    connectivityFailing = true;
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
private static void failOnNotifyTransactionsRead() {
    drGrid().compute().broadcast(new ActiveStoreIgniteRunnable() {
        @Inject
        private transient DoubleLeadService lead;

        @Override protected void runInjected() {
            lead.setWorkInstead(new LeadServiceAdapter() {
                @Override
                public LeadResponse notifyTransactionsRead(UUID consumerId, List<TransactionMetadata> metadatas) {
                    throw new IgniteException("planned fail");
                }
            });
        }
    });
    drGrid().services().cancel(LeadService.SERVICE_NAME);
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * Convert random {@code count} samples from MNIST dataset from two files (images and labels) into libsvm format.
 *
 * @param imagesPath Path to the file with images.
 * @param labelsPath Path to the file with labels.
 * @param outPath Path to output path.
 * @param rnd Random numbers generator.
 * @param cnt Count of samples to read.
 * @throws IgniteException In case of exception.
 */
public static void asLIBSVM(String imagesPath, String labelsPath, String outPath, Random rnd, int cnt)
    throws IOException {

    try (FileWriter fos = new FileWriter(outPath)) {
        mnist(imagesPath, labelsPath, rnd, cnt).forEach(vec -> {
            try {
                fos.write((int)vec.get(vec.size() - 1) + " ");

                for (int i = 0; i < vec.size() - 1; i++) {
                    double val = vec.get(i);

                    if (val != 0)
                        fos.write((i + 1) + ":" + val + " ");
                }

                fos.write("\n");

            }
            catch (IOException e) {
                throw new IgniteException("Error while converting to LIBSVM.");
            }
        });
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * Perform deep copy of an object.
 *
 * @param orig Original object.
 * @param <T> Class of original object;
 * @return Deep copy of original object.
 */
@SuppressWarnings({"unchecked"})
public static <T> T copy(T orig) {
    Object obj;

    try {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream out = new ObjectOutputStream(baos);

        out.writeObject(orig);
        out.flush();
        out.close();

        ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()));

        obj = in.readObject();
    }
    catch (IOException | ClassNotFoundException e) {
        throw new IgniteException("Couldn't copy the object.");
    }

    return (T)obj;
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
@Override
public Boolean execute() throws IgniteException {
    try {
        boolean validateXsdResult = XsdValidator.validate(msg.getMsg(), msg.getXsd());
        boolean validateByJs = JSEvaluate.evaluateJs(msg.getMsg(), msg.getJs());

        Boolean result = validateXsdResult && validateByJs;

        ConcurrentMap<Object, Object> nodeLocalMap = ignite.cluster().nodeLocalMap();

        HttpClient client = HttpAuditClient.createHttpClient(nodeLocalMap);

        return HttpAuditClient.sendResult(client, result, msg.getId());

    } catch (Exception err) {
        throw new IgniteException(err);
    }
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * Make deposit into specified account number.
 *
 * @param accountNumber Bank account number.
 * @param amount Amount to deposit.+
 * @throws IgniteException If failed.
 */
private static void deposit(IgniteCache<Integer, BankAccount> cache, int accountNumber,
    double amount) throws IgniteException {
    try (Transaction tx = Ignition.ignite().transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
        BankAccount bankAccount = cache.get(accountNumber);

        // Deposit into account.
        bankAccount.deposit(amount);

        // Store updated account in cache.
        cache.put(accountNumber, bankAccount);

        tx.commit();
    }

    logger.info("");
    logger.info("|--Deposit amount: $" + amount + "--|");
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/**
 * Make withdraw from specified account number.
 *
 * @param accountNumber Bank Account number.
 * @param amount Amount to withdraw.-
 * @throws IgniteException If failed.
 */
private static void withdraw(IgniteCache<Integer, BankAccount> cache, int accountNumber,
    double amount) throws IgniteException {
    try (Transaction tx = Ignition.ignite().transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
        BankAccount bankAccount = cache.get(accountNumber);

        // Deposit into account.
        bankAccount.withdraw(amount);

        // Store updated account in cache.
        cache.put(accountNumber, bankAccount);

        tx.commit();
    }

    logger.info("");
    logger.info("|--Withdraw amount: $" + amount + "--|");
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
@Override
protected Collection<? extends ComputeJob> split(int gridSize, final Integer arg) throws IgniteException {
    Set<ComputeJob> answer = new HashSet<>();
    for (int i = 0; i < arg; i++) {
        final int c = i;
        answer.add(new ComputeJob() {
            private static final long serialVersionUID = 3365213549618276779L;

            @Override
            public Object execute() throws IgniteException {
                return "a" + c;
            }

            @Override
            public void cancel() {
                // nothing
            }
        });
    }
    return answer;
}
 

import org.apache.ignite.IgniteException; //导入依赖的package包/类
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override public TransactionManager create() {
    try {
        Class clazz = Class.forName("com.ibm.tx.jta.impl.TranManagerSet");

        Method m = clazz.getMethod("instance", (Class[])null);

        TransactionManager tranMgr = (TransactionManager)m.invoke(null, (Object[])null);

        return new WebSphereTransactionManager(tranMgr);
    }
    catch (SecurityException | ClassNotFoundException | IllegalArgumentException | NoSuchMethodException
        | InvocationTargetException | IllegalAccessException e) {
        throw new IgniteException(e);
    }
}