You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by xx...@apache.org on 2020/06/18 02:34:02 UTC
[kylin] 01/02: KYLIN-4577 Throws UnrecognizedPropertyException when
server self discovery enables
This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 6fc39b4d77493ce830555521c95766b5dca1513d
Author: Zhichao Zhang <44...@qq.com>
AuthorDate: Wed Jun 17 18:14:21 2020 +0800
KYLIN-4577 Throws UnrecognizedPropertyException when server self discovery enables
Problem:
When server self discovery enables, it throws UnrecognizedPropertyException on some env, for example: cdh 5.7.
Root cause:
The json of ServiceInstance is incompatible, related to the issue of curator: https://issues.apache.org/jira/browse/CURATOR-394
---
.../common/zookeeper/KylinServerDiscovery.java | 35 +++++++++++++++++++++-
.../common/zookeeper/KylinServerDiscoveryTest.java | 4 +--
.../kylin/job/impl/curator/CuratorScheduler.java | 35 ----------------------
3 files changed, 36 insertions(+), 38 deletions(-)
diff --git a/core-common/src/main/java/org/apache/kylin/common/zookeeper/KylinServerDiscovery.java b/core-common/src/main/java/org/apache/kylin/common/zookeeper/KylinServerDiscovery.java
index 5b67d3e..07f59ad 100644
--- a/core-common/src/main/java/org/apache/kylin/common/zookeeper/KylinServerDiscovery.java
+++ b/core-common/src/main/java/org/apache/kylin/common/zookeeper/KylinServerDiscovery.java
@@ -18,6 +18,7 @@
package org.apache.kylin.common.zookeeper;
+import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.util.LinkedHashMap;
@@ -33,7 +34,7 @@ import org.apache.curator.x.discovery.ServiceCache;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
-import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
+import org.apache.curator.x.discovery.details.InstanceSerializer;
import org.apache.curator.x.discovery.details.ServiceCacheListener;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.StringUtil;
@@ -41,6 +42,9 @@ import org.apache.kylin.common.util.ZKUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -161,4 +165,33 @@ public class KylinServerDiscovery implements Closeable {
IOUtils.closeQuietly(serviceCache);
IOUtils.closeQuietly(serviceDiscovery);
}
+
+ static class JsonInstanceSerializer<T> implements InstanceSerializer<T> {
+ private final ObjectMapper mapper;
+ private final Class<T> payloadClass;
+ private final JavaType type;
+
+ JsonInstanceSerializer(Class<T> payloadClass) {
+ this.payloadClass = payloadClass;
+ this.mapper = new ObjectMapper();
+
+ // to bypass https://issues.apache.org/jira/browse/CURATOR-394
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+ this.type = this.mapper.getTypeFactory().constructType(ServiceInstance.class);
+ }
+
+ public ServiceInstance<T> deserialize(byte[] bytes) throws Exception {
+ ServiceInstance rawServiceInstance = this.mapper.readValue(bytes, this.type);
+ this.payloadClass.cast(rawServiceInstance.getPayload());
+ return rawServiceInstance;
+ }
+
+ public byte[] serialize(ServiceInstance<T> instance) throws Exception {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ mapper.convertValue(instance.getPayload(), payloadClass);
+ this.mapper.writeValue(out, instance);
+ return out.toByteArray();
+ }
+ }
}
diff --git a/core-common/src/test/java/org/apache/kylin/common/zookeeper/KylinServerDiscoveryTest.java b/core-common/src/test/java/org/apache/kylin/common/zookeeper/KylinServerDiscoveryTest.java
index efcb2ef..77dc673 100644
--- a/core-common/src/test/java/org/apache/kylin/common/zookeeper/KylinServerDiscoveryTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/zookeeper/KylinServerDiscoveryTest.java
@@ -29,7 +29,6 @@ import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
-import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.apache.kylin.common.util.LocalFileMetadataTestCase;
import org.apache.kylin.common.util.ZKUtil;
import org.junit.After;
@@ -77,7 +76,8 @@ public class KylinServerDiscoveryTest extends LocalFileMetadataTestCase {
CuratorFramework curatorClient = null;
try {
String servicePath = KylinServerDiscovery.SERVICE_PATH;
- final JsonInstanceSerializer<LinkedHashMap> serializer = new JsonInstanceSerializer<>(LinkedHashMap.class);
+ final KylinServerDiscovery.JsonInstanceSerializer<LinkedHashMap> serializer =
+ new KylinServerDiscovery.JsonInstanceSerializer<>(LinkedHashMap.class);
curatorClient = ZKUtil.newZookeeperClient(zkString, new ExponentialBackoffRetry(3000, 3));
serviceDiscovery = ServiceDiscoveryBuilder.builder(LinkedHashMap.class).client(curatorClient)
.basePath(servicePath).serializer(serializer).build();
diff --git a/core-job/src/main/java/org/apache/kylin/job/impl/curator/CuratorScheduler.java b/core-job/src/main/java/org/apache/kylin/job/impl/curator/CuratorScheduler.java
index 7e829ef..225de32 100644
--- a/core-job/src/main/java/org/apache/kylin/job/impl/curator/CuratorScheduler.java
+++ b/core-job/src/main/java/org/apache/kylin/job/impl/curator/CuratorScheduler.java
@@ -18,7 +18,6 @@
package org.apache.kylin.job.impl.curator;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.util.concurrent.Executors;
@@ -27,8 +26,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.io.IOUtils;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.x.discovery.ServiceInstance;
-import org.apache.curator.x.discovery.details.InstanceSerializer;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.ServerMode;
import org.apache.kylin.common.util.ZKUtil;
@@ -40,9 +37,6 @@ import org.apache.kylin.job.lock.JobLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.JavaType;
-import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
public class CuratorScheduler implements Scheduler<AbstractExecutable> {
@@ -156,33 +150,4 @@ public class CuratorScheduler implements Scheduler<AbstractExecutable> {
return jobClient;
}
- static class JsonInstanceSerializer<T> implements InstanceSerializer<T> {
- private final ObjectMapper mapper;
- private final Class<T> payloadClass;
- private final JavaType type;
-
- JsonInstanceSerializer(Class<T> payloadClass) {
- this.payloadClass = payloadClass;
- this.mapper = new ObjectMapper();
-
- // to bypass https://issues.apache.org/jira/browse/CURATOR-394
- mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
-
- this.type = this.mapper.getTypeFactory().constructType(ServiceInstance.class);
- }
-
- public ServiceInstance<T> deserialize(byte[] bytes) throws Exception {
- ServiceInstance rawServiceInstance = this.mapper.readValue(bytes, this.type);
- this.payloadClass.cast(rawServiceInstance.getPayload());
- return rawServiceInstance;
- }
-
- public byte[] serialize(ServiceInstance<T> instance) throws Exception {
- ByteArrayOutputStream out = new ByteArrayOutputStream();
- mapper.convertValue(instance.getPayload(), payloadClass);
- this.mapper.writeValue(out, instance);
- return out.toByteArray();
- }
- }
-
}