IgnitedAtastReamer, в результате чего валидация состояний перегородки не удалось/раздела Счетчики обновления не соответс ⇐ JAVA

[Anonymous]

Каково значение этой регистрации предупреждения в Apache Ignite? < /p>

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

WARN  [sys-#78%IgniteInstance1%] (Log4J2Logger.java:523) Partition states validation has failed for group: Cache1, msg: Partitions update counters are inconsistent for Part 32...
< /code>
Я пытаюсь использовать несколько экземпляров Ignitedatastreamer для предварительной загрузки кэша после того, как мой кластер запустится, как описано здесь, и я собрал этот мини -пример, чтобы воспроизвести то, что я вижу.  Каждый воспламененный/воспламененный/agnitedatastreamer потокочит с уникальным набором ключей в кэш.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);
}
}
}
< /code>
proloadrunnable.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 запускает потоковые данные в кэш вскоре после того, как он обнаруживает, что Node 2 объединяет кластер, а данные, потоковая диапазон, погружает в диапазон, и диапазон, погружая в диапазон, с диапазоном, и в рамках диапазона. /> Если я изменю паузу в конце waitforservernodeStobeaVailable с 100 миллисекунд до 10 секунд, тогда я делаю не , см. В журнале Warn. Это дает первоначальное время PME для завершения до загрузки .
• Аналогичным образом, если я установите паузу обратно на 100 мс и начало узел 1 после Узел 2, то узел 1 -го узела 1. Сценарий.

В моем примере кода я добавил цикл после LoadCache , который программно вызывает метод Ignite Ignite JMX каждые 10 секунд. В сценарии 1 первый вызов iDleverify подтверждает, что счетчики обновления непоследовательны. Чуть спустя я обычно вижу регистрацию, что указывает на то, что PME завершился. И затем, когда второй запуск idleverify вызов, он сообщает, что конфликтов не было найдено. Таким образом, кажется, что эти счетчики обновления разделения в конечном итоге согласованы . Я также заметил, что если я увеличу объем данных, которые загружаются потоки в кэше, то для завершения первого PME требуется , и в результате это может потребоваться много итераций моего Ideleverify , прежде чем он сообщает, что нет конфликтов. API API IgnitedAtastReamer утверждает, что «стример данных не гарантирует согласованность данных до успешного завершения», но, похоже, это противоречит противоречивым счетчикам обновлений, которые я вижу. Есть ли возможность потери данных, когда мы находимся в этом состоянии? И правильно ли, что эти счетчики обновления всегда будут в конечном итоге согласованными? Предполагая, что топология стабильна (возможно, это плохое предположение), это, по -видимому, избегает всей проблемы с непоследовательными счетчиками обновлений.


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