#755 ExtensionDefault代码调整、MsgHolder增加partition

vipshop · Jul 14, 2021 · 6987297 · 6987297
1 parent 8d360ea
commit 6987297
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 3 deletions.
diff --git a/saturn-core/src/main/java/com/vip/saturn/job/executor/SaturnExecutorExtensionDefault.java b/saturn-core/src/main/java/com/vip/saturn/job/executor/SaturnExecutorExtensionDefault.java
@@ -38,7 +38,7 @@ public class SaturnExecutorExtensionDefault extends SaturnExecutorExtension {
 
 	private static Logger log;
 
-	private static final String NAME_VIP_SATURN_LOG_DIR = "VIP_SATURN_LOG_DIR";
+	protected static final String NAME_VIP_SATURN_LOG_DIR = "VIP_SATURN_LOG_DIR";
 
 	public SaturnExecutorExtensionDefault(String executorName, String namespace, ClassLoader executorClassLoader,
 			ClassLoader jobClassLoader) {
@@ -57,15 +57,15 @@ public void initLogDirEnv() {
 		System.setProperty("saturn.log.dir", saturnLogDir); // for logback.xml
 	}
 
-	private static String getEnv(String key, String defaultValue) {
+	protected static String getEnv(String key, String defaultValue) {
 		String v = System.getenv(key);
 		if (v == null || v.isEmpty()) {
 			return defaultValue;
 		}
 		return v;
 	}
 
-	private static String getDefaultLogDir(String executorName) {
+	protected String getDefaultLogDir(String executorName) {
 		return "/apps/logs/saturn/" + System.getProperty("namespace") + "/" + executorName + "-"
 				+ LocalHostService.cachedIpAddress;
 	}

diff --git a/saturn-job-api/src/main/java/com/vip/saturn/job/msg/MsgHolder.java b/saturn-job-api/src/main/java/com/vip/saturn/job/msg/MsgHolder.java
@@ -41,6 +41,9 @@ public class MsgHolder implements Serializable {
 	/** Kafka offset */
 	private long offset;
 
+	/** Kafka partition */
+	private int partition;
+
 	/**
 	 * @deprecated because the String type of payload maybe is not right
 	 */
@@ -58,6 +61,14 @@ public MsgHolder(byte[] payloadBytes, Set<Entry<String, String>> prop, String me
 		this.offset = offset;
 	}
 
+	public MsgHolder(byte[] payloadBytes, Set<Entry<String, String>> prop, String messageId, long offset, int partition) {// NOSONAR
+		this.payloadBytes = payloadBytes;
+		this.prop = prop;
+		this.messageId = messageId;
+		this.offset = offset;
+		this.partition = partition;
+	}
+
 	public MsgHolder(byte[] payloadBytes, Set<Entry<String, String>> prop, String messageId) {// NOSONAR
 		this.payloadBytes = payloadBytes;
 		this.prop = prop;
@@ -112,6 +123,13 @@ public void copyFrom(Object source) {
 				this.offset = (long) res;
 			}
 
+			field = clazz.getDeclaredField("partition");
+			field.setAccessible(true);
+			res = field.get(source);
+			if (res != null) {
+				this.partition = (int) res;
+			}
+
 		} catch (Exception e) {
 			throw new RuntimeException(e);
 		}
@@ -157,4 +175,8 @@ public String getMessageId() {
 	public long getOffset() {
 		return offset;
 	}
+
+	public int getPartition() {
+		return partition;
+	}
 }