#441 fix, ExecuteJobServerOnlineShardingTask createLocalShard occurs …

…ArrayIndexOutOfBoundsException
vipshop · Jun 25, 2018 · 0924b9f · 0924b9f
1 parent 1a3a945
commit 0924b9f
Show file tree

Hide file tree

Showing 2 changed files with 96 additions and 8 deletions.
diff --git a/saturn-it/src/test/java/com/vip/saturn/it/impl/LocalModeIT.java b/saturn-it/src/test/java/com/vip/saturn/it/impl/LocalModeIT.java
@@ -1,18 +1,23 @@
 package com.vip.saturn.it.impl;
 
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.IThrowableProxy;
 import com.vip.saturn.it.AbstractSaturnIT;
 import com.vip.saturn.it.JobType;
 import com.vip.saturn.it.job.SimpleJavaJob;
+import com.vip.saturn.it.utils.LogbackListAppender;
 import com.vip.saturn.job.executor.Main;
 import com.vip.saturn.job.internal.config.JobConfiguration;
+import com.vip.saturn.job.sharding.task.AbstractAsyncShardingTask;
 import org.apache.commons.exec.OS;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.FixMethodOrder;
-import org.junit.Test;
+import org.junit.*;
 import org.junit.runners.MethodSorters;
 
 import java.io.File;
+import java.util.Iterator;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
 
 @FixMethodOrder(MethodSorters.NAME_ASCENDING)
 public class LocalModeIT extends AbstractSaturnIT {
@@ -21,7 +26,6 @@ public class LocalModeIT extends AbstractSaturnIT {
 	@BeforeClass
 	public static void setUp() throws Exception {
 		startSaturnConsoleList(1);
-		startExecutorList(2);
 		File file1 = new File("src/test/resources/script/normal/normal_0.sh");
 		NORMAL_SH_PATH = file1.getAbsolutePath();
 	}
@@ -32,11 +36,19 @@ public static void tearDown() throws Exception {
 		stopSaturnConsoleList();
 	}
 
+	@After
+	public void after() throws Exception {
+		stopExecutorList();
+		SimpleJavaJob.statusMap.clear();
+	}
+
 	@Test
 	public void test_A() throws Exception {
 		if (!OS.isFamilyUnix()) {
 			return;
 		}
+		startExecutorList(2);
+
 		final JobConfiguration jobConfiguration = new JobConfiguration("shLocalModeJob");
 		jobConfiguration.setCron("*/2 * * * * ?");
 		jobConfiguration.setJobType(JobType.SHELL_JOB.toString());
@@ -81,6 +93,8 @@ public boolean docheck() {
 
 	@Test
 	public void test_B() throws Exception {
+		startExecutorList(2);
+
 		int shardCount = 3;
 		String jobName = "javaLocalModeJob";
 
@@ -134,6 +148,8 @@ public boolean docheck() {
 
 	@Test
 	public void test_C_withPreferList() throws Exception {
+		startExecutorList(2);
+
 		int shardCount = 3;
 		String jobName = "javaLocalModeWithPreferListJob";
 
@@ -197,4 +213,72 @@ public boolean docheck() {
 
 		forceRemoveJob(jobName);
 	}
+
+	@Test
+	public void test_D_fixIssue441() throws Exception {
+		LogbackListAppender logbackListAppender = new LogbackListAppender();
+		logbackListAppender.addToLogger(AbstractAsyncShardingTask.class);
+		logbackListAppender.start();
+
+		final int items = 4;
+		final String jobName = "test_D_fixIssue441";
+
+		final JobConfiguration jobConfiguration = new JobConfiguration(jobName);
+		jobConfiguration.setCron("0/1 * * * * ?");
+		jobConfiguration.setJobType(JobType.JAVA_JOB.toString());
+		jobConfiguration.setProcessCountIntervalSeconds(1);
+		jobConfiguration.setJobClass(SimpleJavaJob.class.getCanonicalName());
+		jobConfiguration.setShardingItemParameters("*=0");
+		jobConfiguration.setLocalMode(true);
+
+		addJob(jobConfiguration);
+		Thread.sleep(1000);
+
+		enableJob(jobName);
+		Thread.sleep(1000);
+
+		startExecutorList(items);
+
+		waitForFinish(new FinishCheck() {
+
+			@Override
+			public boolean docheck() {
+				for (int i = 0; i < items; i++) {
+					if (SimpleJavaJob.statusMap.get(jobName + "_" + i) <= 0) {
+						return false;
+					}
+				}
+				return true;
+			}
+
+		}, 30);
+
+		// 下线0,1,2
+		stopExecutor(0);
+		stopExecutor(1);
+		stopExecutor(2);
+
+		// 上线一个executor
+		startOneNewExecutorList();
+
+		List<ILoggingEvent> allLogs = logbackListAppender.getAllLogs();
+		Iterator<ILoggingEvent> iterator = allLogs.iterator();
+		while (iterator.hasNext()) {
+			ILoggingEvent event = iterator.next();
+			if (event != null) {
+				IThrowableProxy throwableProxy = event.getThrowableProxy();
+				if (throwableProxy != null) {
+					assertThat(throwableProxy.getClassName()).isNotEqualTo("java.lang.ArrayIndexOutOfBoundsException");
+				}
+			}
+		}
+
+		disableJob(jobName);
+		Thread.sleep(1000);
+		removeJob(jobName);
+
+		Thread.sleep(1000);
+
+		forceRemoveJob(jobName);
+	}
 }
diff --git a/...ng/src/main/java/com/vip/saturn/job/sharding/task/ExecuteJobServerOnlineShardingTask.java b/...ng/src/main/java/com/vip/saturn/job/sharding/task/ExecuteJobServerOnlineShardingTask.java
@@ -60,9 +60,13 @@ private Shard createLocalShard(List<Executor> lastOnlineExecutorList, int loadLe
 		}
 		int item = 0;
 		if (!itemList.isEmpty()) {
-			boolean[] flags = new boolean[itemList.size() + 1];
-			for (int i = 0; i < itemList.size(); i++) {
-				flags[itemList.get(i)] = true;
+			int itemListSize = itemList.size();
+			boolean[] flags = new boolean[itemListSize + 1];
+			for (int i = 0; i < itemListSize; i++) {
+				Integer itemAlreadyExists = itemList.get(i);
+				if (itemAlreadyExists <= itemListSize) {
+					flags[itemAlreadyExists] = true;
+				}
 			}
 			for (int i = 0; i < flags.length; i++) {
 				if (!flags[i]) {