You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by qi...@apache.org on 2018/04/13 07:04:11 UTC
carbondata git commit: [CARBONDATA-2337] Fix duplicately acquiring
'streaminglock' error when integrating with spark-streaming
Repository: carbondata
Updated Branches:
refs/heads/branch-1.3 da0cb4f6a -> cd13a4512
[CARBONDATA-2337] Fix duplicately acquiring 'streaminglock' error when integrating with spark-streaming
This closes #2167
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/cd13a451
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/cd13a451
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/cd13a451
Branch: refs/heads/branch-1.3
Commit: cd13a45128a4cfbbf86545a33b4eb431dcdfb5f3
Parents: da0cb4f
Author: Zhang Zhichao <44...@qq.com>
Authored: Fri Apr 13 00:52:03 2018 +0800
Committer: QiangCai <qi...@qq.com>
Committed: Fri Apr 13 15:02:02 2018 +0800
----------------------------------------------------------------------
.../streaming/CarbonStreamSparkStreaming.scala | 33 +++-----------------
1 file changed, 4 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cd13a451/streaming/src/main/scala/org/apache/carbondata/streaming/CarbonStreamSparkStreaming.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/carbondata/streaming/CarbonStreamSparkStreaming.scala b/streaming/src/main/scala/org/apache/carbondata/streaming/CarbonStreamSparkStreaming.scala
index 4aa1517..289d4b3 100644
--- a/streaming/src/main/scala/org/apache/carbondata/streaming/CarbonStreamSparkStreaming.scala
+++ b/streaming/src/main/scala/org/apache/carbondata/streaming/CarbonStreamSparkStreaming.scala
@@ -30,7 +30,6 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.Time
import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.locks.{CarbonLockFactory, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
/**
@@ -47,38 +46,16 @@ class CarbonStreamSparkStreamingWriter(val sparkSession: SparkSession,
private var isInitialize: Boolean = false
- private var lock: ICarbonLock = null
private var carbonAppendableStreamSink: Sink = null
/**
- * Acquired the lock for stream table
- */
- def lockStreamTable(): Unit = {
- lock = CarbonLockFactory.getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
- LockUsage.STREAMING_LOCK)
- if (lock.lockWithRetries()) {
- LOGGER.info("Acquired the lock for stream table: " +
- carbonTable.getDatabaseName + "." +
- carbonTable.getTableName)
- } else {
- LOGGER.error("Not able to acquire the lock for stream table:" +
- carbonTable.getDatabaseName + "." + carbonTable.getTableName)
- throw new InterruptedException(
- "Not able to acquire the lock for stream table: " + carbonTable.getDatabaseName + "." +
- carbonTable.getTableName)
- }
- }
-
- /**
* unlock for stream table
*/
def unLockStreamTable(): Unit = {
- if (null != lock) {
- lock.unlock()
- LOGGER.info("unlock for stream table: " +
- carbonTable.getDatabaseName + "." +
- carbonTable.getTableName)
- }
+ StreamSinkFactory.unLock(carbonTable.getTableUniqueName)
+ LOGGER.info("unlock for stream table: " +
+ carbonTable.getDatabaseName + "." +
+ carbonTable.getTableName)
}
def initialize(): Unit = {
@@ -88,8 +65,6 @@ class CarbonStreamSparkStreamingWriter(val sparkSession: SparkSession,
carbonTable,
extraOptions.toMap).asInstanceOf[CarbonAppendableStreamSink]
- lockStreamTable()
-
isInitialize = true
}