IgniteDataStreamer приводит к сбою проверки состояний разделов/несогласованность счетчиков обновлений разделов ⇐ JAVA

[Anonymous]

Каково значение регистрации этого WARN в Apache Ignite?

Код: Выделить всё

WARN  [sys-#78%IgniteInstance1%] (Log4J2Logger.java:523) Partition states validation has failed for group: Cache1, msg: Partitions update counters are inconsistent for Part 32...

Я пытаюсь использовать несколько экземпляров IgniteDataStreamer для предварительной загрузки кэша после запуска моего кластера, как описано здесь, и я собрал этот мини-пример, чтобы воспроизвести то, что я вижу. Каждый IgniteRunnable/IgniteDataStreamer передает в кеш уникальный набор ключей.
По сути, у меня есть эти два класса:
IgniteServerMain.java

Код: Выделить всё

public class IgniteServerMain {
private static final Logger log = LogManager.getLogger();

public static final String CACHE_NAME = "Cache1";
public static final String IGNITE_INSTANCE_NAME = "IgniteInstance1";
private static final String NODE_ID = System.getProperty("NODE_ID");
private static final int SERVER_NODES = 2;

public static void main(String[] args) {
try {
Ignition.start(getIgniteConfiguration());

if ("1".equals(NODE_ID)) {
waitForServerNodesToBeAvailable();
loadCache();
idleVerifyLoop();
}

while (true) {
sleep(10_000);
}
} catch (Exception e) {
log.error("{}", e, e);
System.exit(1);
}
}

private static IgniteConfiguration getIgniteConfiguration() {
CacheConfiguration cacheConfig = new CacheConfiguration()
.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
.setBackups(1)
.setCacheMode(CacheMode.PARTITIONED)
.setName(CACHE_NAME)
.setReadThrough(false)
.setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC);

DataRegionConfiguration defaultDataRegionConfiguration = new DataRegionConfiguration()
.setName("Default_Region")
.setMaxSize(1L * 1024 * 1024 * 1024)
.setInitialSize(1L * 1024 * 1024 * 1024);

DataStorageConfiguration dataStorageConfiguration = new DataStorageConfiguration()
.setDefaultDataRegionConfiguration(defaultDataRegionConfiguration);

TcpCommunicationSpi tcpCommunicationSpi = new TcpCommunicationSpi();

TcpDiscoveryVmIpFinder tcpDiscoveryVmIpFinder = new TcpDiscoveryVmIpFinder()
.setAddresses(List.of("127.0.0.1:47500..47509"));

TcpDiscoverySpi tcpDiscoverySpi = new TcpDiscoverySpi()
.setIpFinder(tcpDiscoveryVmIpFinder);

return new IgniteConfiguration()
.setCacheConfiguration(cacheConfig)
.setCommunicationSpi(tcpCommunicationSpi)
.setDataStorageConfiguration(dataStorageConfiguration)
.setDiscoverySpi(tcpDiscoverySpi)
.setIgniteInstanceName(IGNITE_INSTANCE_NAME)
.setIncludeEventTypes(EventType.EVT_NODE_FAILED);
}

private static void waitForServerNodesToBeAvailable() {
log.info("waiting for server nodes");
Ignite ignite = Ignition.ignite(IGNITE_INSTANCE_NAME);
while (ignite.cluster().forServers().nodes().size() < SERVER_NODES) {
sleep(1);
}

log.info("server nodes are available!");
sleep(100);
}

private static void loadCache() {
Instant start = Instant.now();
log.info("starting {}", CACHE_NAME);

Ignite ignite = Ignition.ignite(IGNITE_INSTANCE_NAME);
IgniteCompute compute = ignite.compute();
List jobs = IntStream.range(0, 100)
.mapToObj(PreloadRunnable::new)
.map(compute::runAsync)
.collect(Collectors.toList());

jobs.forEach(IgniteFuture::get);

Duration duration = Duration.between(start, Instant.now());
log.info("finished: duration {}", duration);

int size = ignite.cache(CACHE_NAME).size();
log.info("cache {}, size {}", CACHE_NAME, size);
}

private static boolean idleVerify() {
Instant start = Instant.now();

MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
ObjectName objectName = mbs.queryMBeans(null, null).stream()
.filter(objectInstance -> objectInstance.toString().contains("name=IdleVerify"))
.map(objectInstance -> objectInstance.getObjectName())
.findFirst()
.orElseThrow(() ->  new IllegalStateException("IdleVerify: MBean not found"));

try {
String result = (String) mbs.invoke(objectName, "invoke",
new Object[] { "", "", "", "", "" },
new String[] {});
log.info("IdleVerify: finished in {}", Duration.between(start, Instant.now()));

String[] resultSplit = result.split("\\R");
Stream.of(resultSplit).forEach(s -> log.info("IdleVerify: {}", s));
return "The check procedure has finished, no conflicts have been found."
.equals(resultSplit[resultSplit.length - 1]);
} catch (Exception e) {
log.error(e.toString(), e);
return false;
}
}

private static void idleVerifyLoop() {
Instant start = Instant.now();

while (!idleVerify()) {
if (Duration.between(start, Instant.now()).getSeconds() > 120) {
log.error("IdleVerifyLoop: exiting, there are still conflicts after 2 minutes of waiting");
return;
}
sleep(10_000);
}

Duration duration = Duration.between(start, Instant.now());
log.info("IdleVerifyLoop: finished in {}", duration);
}

private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
log.error(e.toString(), e);
Thread.interrupted();
throw new RuntimeException(e);
}
}
}

PreloadRunnable.java

Код: Выделить всё

public class PreloadRunnable implements IgniteRunnable {
private static final long serialVersionUID = 1L;

private final int jobId;
private final Random random = new Random();

public PreloadRunnable(int jobId) {
this.jobId = jobId;
}

@Override
public void run() {
try (IgniteDataStreamer streamer = Ignition
.ignite(IgniteServerMain.IGNITE_INSTANCE_NAME)
.dataStreamer(IgniteServerMain.CACHE_NAME)) {

for (int v = 0; v < 10_000; v++) {
char randomLetter = (char) ('A' + random.nextInt(26));
String k = randomLetter + "-" + String.format("%06d", jobId) + "-" + String.format("%06d", v);
streamer.addData(k, v);
}
}
}
}

Несколько сценариев:

• Я наблюдаю регистрацию WARN всякий раз, когда запускаю узел 1 перед узлом 2. В этом сценарии узел 1 начинает потоковую передачу данных в кэш вскоре после того, как обнаруживает, что узел 2 присоединился к кластеру, и потоковая передача данных чередуется с обменом картами разделов.
• Если я изменю паузу в конце waitForServerNodesToBeAvailable со 100 миллисекунд на 10 секунд, то я не увижу журнал WARN. Это дает начальному PME время для завершения до loadCache.
• Аналогично, если я установлю паузу обратно на 100 мс и запущу узел 1 после узла 2, то метод Ignition.start узла 1 не вернется до тех пор, пока после завершится первоначальный PME, и поэтому я также не вижу предупреждение WARN вход в систему в этом сценарии.

В моем примере кода я добавил цикл после loadCache, который программно вызывает метод JMX Ignite IdleVerify каждые 10 секунд. В сценарии 1 первый вызов IdleVerify подтверждает, что счетчики обновлений несогласованы. Чуть позже я обычно вижу журналирование, которое указывает на завершение PME. А затем, когда запускается второй вызов IdleVerify, он сообщает, что конфликтов не обнаружено. Таким образом, кажется, что эти счетчики обновлений разделов в конечном итоге согласованы. Я также заметил, что если я увеличиваю объем данных, которые loadCache передает в кеш, то для завершения начального PME потребуется больше, и в результате может потребоваться много итераций моего цикла IdleVerify, прежде чем он сообщит об отсутствии конфликтов.
Вопросы:

• Что здесь происходит? API IgniteDataStreamer утверждает, что «стример данных не гарантирует [...] согласованность данных до успешного завершения», но это, похоже, противоречит противоречивым счетчикам обновлений, которые я вижу.
• Каково значение этих противоречивых счетчиков обновлений? Есть ли вероятность потери данных, когда мы находимся в таком состоянии? И правильно ли, что эти счетчики обновлений всегда будут в конечном итоге согласованными?
• Есть ли лучший способ для узла 1 дождаться подключения других узлов и гарантировать, что начальный PME завершится до того, как он начнет потоковую передачу данных в кеш? Если предположить, что топология стабильна (возможно, это неверное предположение), это, похоже, позволяет избежать всей проблемы с несогласованными счетчиками обновлений.

Подробнее здесь: https://stackoverflow.com/questions/796 ... -partition