我在aws emr中使用在apache flink上运行的aws keyspace（cassandra3.11.2）。查询引发异常。在aws lambda上使用的相同代码也有相同的异常nohost。我做错了什么？

String query = "INSERT INTO TEST (field1, field2) VALUES(?, ?)";
PreparedStatement prepared = CassandraConnector.prepare(query);
int i = 0;
BoundStatement bound = prepared.bind().setString(i++, "Field1").setString(i++, "Field2")
                    .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
ResultSet rs = CassandraConnector.execute(bound);

at com.datastax.oss.driver.api.core.NoNodeAvailableException.copy(NoNodeAvailableException.java:40)
 at com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures.getUninterruptibly(CompletableFutures.java:149)
 at com.datastax.oss.driver.internal.core.cql.CqlRequestSyncProcessor.process(CqlRequestSyncProcessor.java:53)
 at com.datastax.oss.driver.internal.core.cql.CqlRequestSyncProcessor.process(CqlRequestSyncProcessor.java:30)
 at com.datastax.oss.driver.internal.core.session.DefaultSession.execute(DefaultSession.java:230)
 at com.datastax.oss.driver.api.core.cql.SyncCqlSession.execute(SyncCqlSession.java:53)
 at com.test.manager.connectors.CassandraConnector.execute(CassandraConnector.java:16)
 at com.test.repository.impl.BackupRepositoryImpl.insert(BackupRepositoryImpl.java:36)
 at com.test.service.impl.BackupServiceImpl.insert(BackupServiceImpl.java:18)
 at com.test.flink.function.AsyncBackupFunction.processMessage(AsyncBackupFunction.java:78)
 at com.test.flink.function.AsyncBackupFunction.lambda$asyncInvoke$0(AsyncBackupFunction.java:35)
 at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
 at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1596)
 at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
 at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
 at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
 at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

这是我的密码：
java：因为init preparedstatement的开销很大，所以我缓存了这个。

public class CassandraConnector {
    private static final ConcurrentHashMap<String, PreparedStatement> preparedStatementCache = new ConcurrentHashMap<String, PreparedStatement>();

    public static ResultSet execute(BoundStatement bound) {
        CqlSession session = CassandraManager.getSessionInstance();
        return session.execute(bound);
    }

    public static ResultSet execute(String query) {
        CqlSession session = CassandraManager.getSessionInstance();
        return session.execute(query);
    }

    public static PreparedStatement prepare(String query) {
        PreparedStatement result = preparedStatementCache.get(query);
        if (result == null) {
            CqlSession session = CassandraManager.getSessionInstance();
            result = session.prepare(query);
            preparedStatementCache.putIfAbsent(query, result);
        }

        return result;
    }
}

java：我正在为会话对象使用单例双重检查锁定。

public class CassandraManager {
    private static final Logger logger = LoggerFactory.getLogger(CassandraManager.class);
    private static final String SSL_CASSANDRA_PASSWORD = "password";
    private static volatile CqlSession session;

    static {
        try {
            initSession();
        } catch (Exception e) {
            logger.error("Error CassandraManager getSessionInstance", e);
        }
    }

    private static void initSession() {
        List<InetSocketAddress> contactPoints = Collections.singletonList(InetSocketAddress.createUnresolved(
                "cassandra.ap-southeast-1.amazonaws.com", 9142));
        DriverConfigLoader loader = DriverConfigLoader.fromClasspath("application.conf");

        Long start = BaseHelper.getTime();
        session = CqlSession.builder().addContactPoints(contactPoints).withConfigLoader(loader)
                .withAuthCredentials(AppUtil.getProperty("cassandra.username"),
                        AppUtil.getProperty("cassandra.password"))
                .withSslContext(getSSLContext()).withLocalDatacenter("ap-southeast-1")
                .withKeyspace(AppUtil.getProperty("cassandra.keyspace")).build();
        logger.info("End connect: " + (new Date().getTime() - start));

    }

    public static CqlSession getSessionInstance() {
        if (session == null || session.isClosed()) {
            synchronized (CassandraManager.class) {
                if (session == null || session.isClosed()) {
                    initSession();
                }
            }
        }

        return session;
    }

    public static SSLContext getSSLContext() {
        InputStream in = null;
        try {
            KeyStore ks = KeyStore.getInstance("JKS");
            in = CassandraManager.class.getClassLoader().getResourceAsStream("cassandra_truststore.jks");
            ks.load(in, SSL_CASSANDRA_PASSWORD.toCharArray());
            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
            tmf.init(ks);

            SSLContext ctx = SSLContext.getInstance("TLS");
            ctx.init(null, tmf.getTrustManagers(), null);
            return ctx;
        } catch (Exception e) {
            logger.error("Error CassandraConnector getSSLContext", e);
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    logger.error("", e);
                }
            }
        }

        return null;
    }
}

应用程序.conf

datastax-java-driver {
  basic.request {
    timeout = 5 seconds
    consistency = LOCAL_ONE
  }
  advanced.connection {
    max-requests-per-connection = 1024
    pool {
      local.size = 1
      remote.size = 1
    }
  }
  advanced.reconnect-on-init = true
  advanced.reconnection-policy {
    class = ExponentialReconnectionPolicy
    base-delay = 1 second
    max-delay = 60 seconds
  }
  advanced.retry-policy {
    class = DefaultRetryPolicy
  }
  advanced.protocol {
    version = V4
  }

  advanced.heartbeat {
    interval = 30 seconds
    timeout = 1 second
  }

  advanced.session-leak.threshold = 8
  advanced.metadata.token-map.enabled = false
}