Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
A
atlas
Project
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
dataplatform
atlas
Commits
8903c9a6
Commit
8903c9a6
authored
6 years ago
by
Ashutosh Mestry
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
ATLAS-2862: Incremental Export now uses request context to determine change marker.
parent
708e4865
master
No related merge requests found
Show whitespace changes
Inline
Side-by-side
Showing
11 changed files
with
87 additions
and
101 deletions
+87
-101
AtlasExportRequest.java
...ava/org/apache/atlas/model/impexp/AtlasExportRequest.java
+44
-1
AtlasExportResult.java
...java/org/apache/atlas/model/impexp/AtlasExportResult.java
+10
-32
AtlasImportResult.java
...java/org/apache/atlas/model/impexp/AtlasImportResult.java
+1
-1
AuditsWriter.java
...java/org/apache/atlas/repository/impexp/AuditsWriter.java
+2
-2
ExportService.java
...ava/org/apache/atlas/repository/impexp/ExportService.java
+20
-57
ZipSink.java
...main/java/org/apache/atlas/repository/impexp/ZipSink.java
+3
-2
ExportIncrementalTest.java
...apache/atlas/repository/impexp/ExportIncrementalTest.java
+3
-3
ImportTransformsShaperTest.java
...e/atlas/repository/impexp/ImportTransformsShaperTest.java
+1
-0
ReplicationEntityAttributeTest.java
...las/repository/impexp/ReplicationEntityAttributeTest.java
+1
-1
ZipSinkTest.java
.../java/org/apache/atlas/repository/impexp/ZipSinkTest.java
+1
-1
export-incremental.json
.../resources/json/stocksDB-Entities/export-incremental.json
+1
-1
No files found.
intg/src/main/java/org/apache/atlas/model/impexp/AtlasExportRequest.java
View file @
8903c9a6
...
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
...
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import
com.fasterxml.jackson.databind.annotation.JsonSerialize
;
import
com.fasterxml.jackson.databind.annotation.JsonSerialize
;
import
org.apache.atlas.model.instance.AtlasObjectId
;
import
org.apache.atlas.model.instance.AtlasObjectId
;
import
org.apache.atlas.model.typedef.AtlasBaseTypeDef
;
import
org.apache.atlas.model.typedef.AtlasBaseTypeDef
;
import
org.apache.commons.collections.MapUtils
;
import
javax.xml.bind.annotation.XmlAccessType
;
import
javax.xml.bind.annotation.XmlAccessType
;
import
javax.xml.bind.annotation.XmlAccessorType
;
import
javax.xml.bind.annotation.XmlAccessorType
;
...
@@ -53,7 +54,7 @@ public class AtlasExportRequest implements Serializable {
...
@@ -53,7 +54,7 @@ public class AtlasExportRequest implements Serializable {
public
static
final
String
FETCH_TYPE_FULL
=
"full"
;
public
static
final
String
FETCH_TYPE_FULL
=
"full"
;
public
static
final
String
FETCH_TYPE_CONNECTED
=
"connected"
;
public
static
final
String
FETCH_TYPE_CONNECTED
=
"connected"
;
public
static
final
String
FETCH_TYPE_INCREMENTAL
=
"incremental"
;
public
static
final
String
FETCH_TYPE_INCREMENTAL
=
"incremental"
;
public
static
final
String
FETCH_TYPE_INCREMENTAL_
FROM_TIME
=
"fromTime
"
;
public
static
final
String
FETCH_TYPE_INCREMENTAL_
CHANGE_MARKER
=
"changeMarker
"
;
public
static
final
String
MATCH_TYPE_STARTS_WITH
=
"startsWith"
;
public
static
final
String
MATCH_TYPE_STARTS_WITH
=
"startsWith"
;
public
static
final
String
MATCH_TYPE_ENDS_WITH
=
"endsWith"
;
public
static
final
String
MATCH_TYPE_ENDS_WITH
=
"endsWith"
;
public
static
final
String
MATCH_TYPE_CONTAINS
=
"contains"
;
public
static
final
String
MATCH_TYPE_CONTAINS
=
"contains"
;
...
@@ -79,6 +80,48 @@ public class AtlasExportRequest implements Serializable {
...
@@ -79,6 +80,48 @@ public class AtlasExportRequest implements Serializable {
this
.
options
=
options
;
this
.
options
=
options
;
}
}
public
String
getMatchTypeOptionValue
()
{
String
matchType
=
null
;
if
(
MapUtils
.
isNotEmpty
(
getOptions
()))
{
if
(
getOptions
().
get
(
OPTION_ATTR_MATCH_TYPE
)
!=
null
)
{
matchType
=
getOptions
().
get
(
OPTION_ATTR_MATCH_TYPE
).
toString
();
}
}
return
matchType
;
}
public
String
getFetchTypeOptionValue
()
{
if
(
getOptions
()
==
null
||
!
getOptions
().
containsKey
(
OPTION_FETCH_TYPE
))
{
return
FETCH_TYPE_FULL
;
}
Object
o
=
getOptions
().
get
(
OPTION_FETCH_TYPE
);
if
(
o
instanceof
String
)
{
return
(
String
)
o
;
}
return
FETCH_TYPE_FULL
;
}
public
boolean
getSkipLineageOptionValue
()
{
if
(!
getOptions
().
containsKey
(
AtlasExportRequest
.
OPTION_SKIP_LINEAGE
))
{
return
false
;
}
Object
o
=
getOptions
().
get
(
AtlasExportRequest
.
OPTION_SKIP_LINEAGE
);
if
(
o
instanceof
String
)
{
return
Boolean
.
parseBoolean
((
String
)
o
);
}
if
(
o
instanceof
Boolean
)
{
return
(
Boolean
)
o
;
}
return
false
;
}
public
StringBuilder
toString
(
StringBuilder
sb
)
{
public
StringBuilder
toString
(
StringBuilder
sb
)
{
if
(
sb
==
null
)
{
if
(
sb
==
null
)
{
sb
=
new
StringBuilder
();
sb
=
new
StringBuilder
();
...
...
This diff is collapsed.
Click to expand it.
intg/src/main/java/org/apache/atlas/model/impexp/AtlasExportResult.java
View file @
8903c9a6
...
@@ -62,14 +62,15 @@ public class AtlasExportResult implements Serializable {
...
@@ -62,14 +62,15 @@ public class AtlasExportResult implements Serializable {
private
AtlasExportData
data
;
private
AtlasExportData
data
;
private
OperationStatus
operationStatus
;
private
OperationStatus
operationStatus
;
private
String
sourceClusterName
;
private
String
sourceClusterName
;
private
long
lastModifiedTimestamp
;
private
long
changeMarker
;
public
AtlasExportResult
()
{
public
AtlasExportResult
()
{
this
(
null
,
null
,
null
,
null
,
System
.
currentTimeMillis
());
this
(
null
,
null
,
null
,
null
,
System
.
currentTimeMillis
()
,
0L
);
}
}
public
AtlasExportResult
(
AtlasExportRequest
request
,
public
AtlasExportResult
(
AtlasExportRequest
request
,
String
userName
,
String
clientIpAddress
,
String
hostName
,
long
timeStamp
)
{
String
userName
,
String
clientIpAddress
,
String
hostName
,
long
timeStamp
,
long
changeMarker
)
{
this
.
request
=
request
;
this
.
request
=
request
;
this
.
userName
=
userName
;
this
.
userName
=
userName
;
this
.
clientIpAddress
=
clientIpAddress
;
this
.
clientIpAddress
=
clientIpAddress
;
...
@@ -78,6 +79,7 @@ public class AtlasExportResult implements Serializable {
...
@@ -78,6 +79,7 @@ public class AtlasExportResult implements Serializable {
this
.
metrics
=
new
HashMap
<>();
this
.
metrics
=
new
HashMap
<>();
this
.
operationStatus
=
OperationStatus
.
FAIL
;
this
.
operationStatus
=
OperationStatus
.
FAIL
;
this
.
data
=
new
AtlasExportData
();
this
.
data
=
new
AtlasExportData
();
this
.
changeMarker
=
changeMarker
;
}
}
public
AtlasExportRequest
getRequest
()
{
public
AtlasExportRequest
getRequest
()
{
...
@@ -136,12 +138,12 @@ public class AtlasExportResult implements Serializable {
...
@@ -136,12 +138,12 @@ public class AtlasExportResult implements Serializable {
this
.
data
=
data
;
this
.
data
=
data
;
}
}
public
void
set
LastModifiedTimestamp
(
long
lastModifiedTimestamp
)
{
public
void
set
ChangeMarker
(
long
changeMarker
)
{
this
.
lastModifiedTimestamp
=
lastModifiedTimestamp
;
this
.
changeMarker
=
changeMarker
;
}
}
public
long
get
LastModifiedTimestamp
()
{
public
long
get
ChangeMarker
()
{
return
this
.
lastModifiedTimestamp
;
return
this
.
changeMarker
;
}
}
public
OperationStatus
getOperationStatus
()
{
public
OperationStatus
getOperationStatus
()
{
...
@@ -173,22 +175,6 @@ public class AtlasExportResult implements Serializable {
...
@@ -173,22 +175,6 @@ public class AtlasExportResult implements Serializable {
metrics
.
put
(
key
,
currentValue
+
incrementBy
);
metrics
.
put
(
key
,
currentValue
+
incrementBy
);
}
}
public
AtlasExportResult
shallowCopy
()
{
AtlasExportResult
result
=
new
AtlasExportResult
();
result
.
setRequest
(
getRequest
());
result
.
setUserName
(
getUserName
());
result
.
setClientIpAddress
(
getClientIpAddress
());
result
.
setHostName
(
getHostName
());
result
.
setTimeStamp
(
getTimeStamp
());
result
.
setMetrics
(
getMetrics
());
result
.
setOperationStatus
(
getOperationStatus
());
result
.
setSourceClusterName
(
getSourceClusterName
());
result
.
setLastModifiedTimestamp
(
getLastModifiedTimestamp
());
return
result
;
}
public
StringBuilder
toString
(
StringBuilder
sb
)
{
public
StringBuilder
toString
(
StringBuilder
sb
)
{
if
(
sb
==
null
)
{
if
(
sb
==
null
)
{
sb
=
new
StringBuilder
();
sb
=
new
StringBuilder
();
...
@@ -199,14 +185,13 @@ public class AtlasExportResult implements Serializable {
...
@@ -199,14 +185,13 @@ public class AtlasExportResult implements Serializable {
sb
.
append
(
", userName='"
).
append
(
userName
).
append
(
"'"
);
sb
.
append
(
", userName='"
).
append
(
userName
).
append
(
"'"
);
sb
.
append
(
", clientIpAddress='"
).
append
(
clientIpAddress
).
append
(
"'"
);
sb
.
append
(
", clientIpAddress='"
).
append
(
clientIpAddress
).
append
(
"'"
);
sb
.
append
(
", hostName='"
).
append
(
hostName
).
append
(
"'"
);
sb
.
append
(
", hostName='"
).
append
(
hostName
).
append
(
"'"
);
sb
.
append
(
",
lastModifiedTimestamp='"
).
append
(
lastModifiedTimestamp
).
append
(
"'"
);
sb
.
append
(
",
changeMarker='"
).
append
(
changeMarker
).
append
(
"'"
);
sb
.
append
(
", sourceCluster='"
).
append
(
sourceClusterName
).
append
(
"'"
);
sb
.
append
(
", sourceCluster='"
).
append
(
sourceClusterName
).
append
(
"'"
);
sb
.
append
(
", timeStamp='"
).
append
(
timeStamp
).
append
(
"'"
);
sb
.
append
(
", timeStamp='"
).
append
(
timeStamp
).
append
(
"'"
);
sb
.
append
(
", metrics={"
);
sb
.
append
(
", metrics={"
);
AtlasBaseTypeDef
.
dumpObjects
(
metrics
,
sb
);
AtlasBaseTypeDef
.
dumpObjects
(
metrics
,
sb
);
sb
.
append
(
"}"
);
sb
.
append
(
"}"
);
sb
.
append
(
", data='"
).
append
(
data
).
append
(
"'"
);
sb
.
append
(
", operationStatus='"
).
append
(
operationStatus
).
append
(
"'"
);
sb
.
append
(
", operationStatus='"
).
append
(
operationStatus
).
append
(
"'"
);
sb
.
append
(
"}"
);
sb
.
append
(
"}"
);
...
@@ -237,13 +222,11 @@ public class AtlasExportResult implements Serializable {
...
@@ -237,13 +222,11 @@ public class AtlasExportResult implements Serializable {
private
static
final
long
serialVersionUID
=
1L
;
private
static
final
long
serialVersionUID
=
1L
;
private
AtlasTypesDef
typesDef
;
private
AtlasTypesDef
typesDef
;
private
Map
<
String
,
AtlasEntity
>
entities
;
private
List
<
String
>
entityCreationOrder
;
private
List
<
String
>
entityCreationOrder
;
public
AtlasExportData
()
{
public
AtlasExportData
()
{
typesDef
=
new
AtlasTypesDef
();
typesDef
=
new
AtlasTypesDef
();
entities
=
new
HashMap
<>();
entityCreationOrder
=
new
ArrayList
<>();
entityCreationOrder
=
new
ArrayList
<>();
}
}
...
@@ -251,10 +234,6 @@ public class AtlasExportResult implements Serializable {
...
@@ -251,10 +234,6 @@ public class AtlasExportResult implements Serializable {
public
void
setTypesDef
(
AtlasTypesDef
typesDef
)
{
this
.
typesDef
=
typesDef
;
}
public
void
setTypesDef
(
AtlasTypesDef
typesDef
)
{
this
.
typesDef
=
typesDef
;
}
public
Map
<
String
,
AtlasEntity
>
getEntities
()
{
return
entities
;
}
public
void
setEntities
(
Map
<
String
,
AtlasEntity
>
entities
)
{
this
.
entities
=
entities
;
}
public
List
<
String
>
getEntityCreationOrder
()
{
return
entityCreationOrder
;
}
public
List
<
String
>
getEntityCreationOrder
()
{
return
entityCreationOrder
;
}
public
void
setEntityCreationOrder
(
List
<
String
>
entityCreationOrder
)
{
this
.
entityCreationOrder
=
entityCreationOrder
;
}
public
void
setEntityCreationOrder
(
List
<
String
>
entityCreationOrder
)
{
this
.
entityCreationOrder
=
entityCreationOrder
;
}
...
@@ -267,7 +246,6 @@ public class AtlasExportResult implements Serializable {
...
@@ -267,7 +246,6 @@ public class AtlasExportResult implements Serializable {
sb
.
append
(
"AtlasExportData {"
);
sb
.
append
(
"AtlasExportData {"
);
sb
.
append
(
", typesDef={"
).
append
(
typesDef
).
append
(
"}"
);
sb
.
append
(
", typesDef={"
).
append
(
typesDef
).
append
(
"}"
);
sb
.
append
(
", entities={"
);
sb
.
append
(
", entities={"
);
AtlasBaseTypeDef
.
dumpObjects
(
entities
,
sb
);
sb
.
append
(
"}"
);
sb
.
append
(
"}"
);
sb
.
append
(
", entityCreationOrder={"
);
sb
.
append
(
", entityCreationOrder={"
);
AtlasBaseTypeDef
.
dumpObjects
(
entityCreationOrder
,
sb
);
AtlasBaseTypeDef
.
dumpObjects
(
entityCreationOrder
,
sb
);
...
...
This diff is collapsed.
Click to expand it.
intg/src/main/java/org/apache/atlas/model/impexp/AtlasImportResult.java
View file @
8903c9a6
...
@@ -149,7 +149,7 @@ public class AtlasImportResult {
...
@@ -149,7 +149,7 @@ public class AtlasImportResult {
}
}
public
void
setExportResult
(
AtlasExportResult
exportResult
)
{
public
void
setExportResult
(
AtlasExportResult
exportResult
)
{
this
.
exportResultWithoutData
=
exportResult
.
shallowCopy
()
;
this
.
exportResultWithoutData
=
exportResult
;
}
}
public
StringBuilder
toString
(
StringBuilder
sb
)
{
public
StringBuilder
toString
(
StringBuilder
sb
)
{
...
...
This diff is collapsed.
Click to expand it.
repository/src/main/java/org/apache/atlas/repository/impexp/AuditsWriter.java
View file @
8903c9a6
...
@@ -144,7 +144,7 @@ public class AuditsWriter {
...
@@ -144,7 +144,7 @@ public class AuditsWriter {
}
}
updateReplicationAttribute
(
replicationOptionState
,
targetServerName
,
updateReplicationAttribute
(
replicationOptionState
,
targetServerName
,
entityGuids
,
Constants
.
ATTR_NAME_REPLICATED_TO
,
result
.
get
LastModifiedTimestamp
());
entityGuids
,
Constants
.
ATTR_NAME_REPLICATED_TO
,
result
.
get
ChangeMarker
());
}
}
private
void
saveServers
()
throws
AtlasBaseException
{
private
void
saveServers
()
throws
AtlasBaseException
{
...
@@ -182,7 +182,7 @@ public class AuditsWriter {
...
@@ -182,7 +182,7 @@ public class AuditsWriter {
}
}
updateReplicationAttribute
(
replicationOptionState
,
this
.
sourceServerName
,
entityGuids
,
updateReplicationAttribute
(
replicationOptionState
,
this
.
sourceServerName
,
entityGuids
,
Constants
.
ATTR_NAME_REPLICATED_FROM
,
result
.
getExportResult
().
get
LastModifiedTimestamp
());
Constants
.
ATTR_NAME_REPLICATED_FROM
,
result
.
getExportResult
().
get
ChangeMarker
());
}
}
private
void
saveServers
()
throws
AtlasBaseException
{
private
void
saveServers
()
throws
AtlasBaseException
{
...
...
This diff is collapsed.
Click to expand it.
repository/src/main/java/org/apache/atlas/repository/impexp/ExportService.java
View file @
8903c9a6
...
@@ -21,6 +21,7 @@ import com.google.common.annotations.VisibleForTesting;
...
@@ -21,6 +21,7 @@ import com.google.common.annotations.VisibleForTesting;
import
org.apache.atlas.AtlasErrorCode
;
import
org.apache.atlas.AtlasErrorCode
;
import
org.apache.atlas.AtlasException
;
import
org.apache.atlas.AtlasException
;
import
org.apache.atlas.AtlasServiceException
;
import
org.apache.atlas.AtlasServiceException
;
import
org.apache.atlas.RequestContext
;
import
org.apache.atlas.exception.AtlasBaseException
;
import
org.apache.atlas.exception.AtlasBaseException
;
import
org.apache.atlas.model.TypeCategory
;
import
org.apache.atlas.model.TypeCategory
;
import
org.apache.atlas.model.impexp.AtlasExportRequest
;
import
org.apache.atlas.model.impexp.AtlasExportRequest
;
...
@@ -93,7 +94,9 @@ public class ExportService {
...
@@ -93,7 +94,9 @@ public class ExportService {
public
AtlasExportResult
run
(
ZipSink
exportSink
,
AtlasExportRequest
request
,
String
userName
,
String
hostName
,
public
AtlasExportResult
run
(
ZipSink
exportSink
,
AtlasExportRequest
request
,
String
userName
,
String
hostName
,
String
requestingIP
)
throws
AtlasBaseException
{
String
requestingIP
)
throws
AtlasBaseException
{
long
startTime
=
System
.
currentTimeMillis
();
long
startTime
=
System
.
currentTimeMillis
();
AtlasExportResult
result
=
new
AtlasExportResult
(
request
,
userName
,
requestingIP
,
hostName
,
startTime
);
AtlasExportResult
result
=
new
AtlasExportResult
(
request
,
userName
,
requestingIP
,
hostName
,
startTime
,
getCurrentChangeMarker
());
ExportContext
context
=
new
ExportContext
(
atlasGraph
,
result
,
exportSink
);
ExportContext
context
=
new
ExportContext
(
atlasGraph
,
result
,
exportSink
);
exportTypeProcessor
=
new
ExportTypeProcessor
(
typeRegistry
,
context
);
exportTypeProcessor
=
new
ExportTypeProcessor
(
typeRegistry
,
context
);
...
@@ -117,6 +120,10 @@ public class ExportService {
...
@@ -117,6 +120,10 @@ public class ExportService {
return
context
.
result
;
return
context
.
result
;
}
}
private
long
getCurrentChangeMarker
()
{
return
RequestContext
.
earliestActiveRequestTime
();
}
private
void
updateSinkWithOperationMetrics
(
String
userName
,
ExportContext
context
,
private
void
updateSinkWithOperationMetrics
(
String
userName
,
ExportContext
context
,
AtlasExportResult
.
OperationStatus
[]
statuses
,
AtlasExportResult
.
OperationStatus
[]
statuses
,
long
startTime
,
long
endTime
)
throws
AtlasBaseException
{
long
startTime
,
long
endTime
)
throws
AtlasBaseException
{
...
@@ -125,7 +132,6 @@ public class ExportService {
...
@@ -125,7 +132,6 @@ public class ExportService {
context
.
result
.
getData
().
getEntityCreationOrder
().
addAll
(
context
.
lineageProcessed
);
context
.
result
.
getData
().
getEntityCreationOrder
().
addAll
(
context
.
lineageProcessed
);
context
.
sink
.
setExportOrder
(
context
.
result
.
getData
().
getEntityCreationOrder
());
context
.
sink
.
setExportOrder
(
context
.
result
.
getData
().
getEntityCreationOrder
());
context
.
sink
.
setTypesDef
(
context
.
result
.
getData
().
getTypesDef
());
context
.
sink
.
setTypesDef
(
context
.
result
.
getData
().
getTypesDef
());
context
.
result
.
setLastModifiedTimestamp
(
context
.
newestLastModifiedTimestamp
);
context
.
result
.
setOperationStatus
(
getOverallOperationStatus
(
statuses
));
context
.
result
.
setOperationStatus
(
getOverallOperationStatus
(
statuses
));
context
.
result
.
incrementMeticsCounter
(
"duration"
,
duration
);
context
.
result
.
incrementMeticsCounter
(
"duration"
,
duration
);
auditsWriter
.
write
(
userName
,
context
.
result
,
startTime
,
endTime
,
context
.
result
.
getData
().
getEntityCreationOrder
());
auditsWriter
.
write
(
userName
,
context
.
result
,
startTime
,
endTime
,
context
.
result
.
getData
().
getEntityCreationOrder
());
...
@@ -362,7 +368,7 @@ public class ExportService {
...
@@ -362,7 +368,7 @@ public class ExportService {
debugLog
(
"<== processEntity({})"
,
guid
);
debugLog
(
"<== processEntity({})"
,
guid
);
}
}
private
void
getConntedEntitiesBasedOnOption
(
AtlasEntity
entity
,
ExportContext
context
,
TraversalDirection
direction
)
throws
AtlasBaseException
{
private
void
getConntedEntitiesBasedOnOption
(
AtlasEntity
entity
,
ExportContext
context
,
TraversalDirection
direction
)
{
switch
(
context
.
fetchType
)
{
switch
(
context
.
fetchType
)
{
case
CONNECTED:
case
CONNECTED:
getEntityGuidsForConnectedFetch
(
entity
,
context
,
direction
);
getEntityGuidsForConnectedFetch
(
entity
,
context
,
direction
);
...
@@ -375,7 +381,7 @@ public class ExportService {
...
@@ -375,7 +381,7 @@ public class ExportService {
}
}
}
}
private
void
getEntityGuidsForConnectedFetch
(
AtlasEntity
entity
,
ExportContext
context
,
TraversalDirection
direction
)
throws
AtlasBaseException
{
private
void
getEntityGuidsForConnectedFetch
(
AtlasEntity
entity
,
ExportContext
context
,
TraversalDirection
direction
)
{
if
(
direction
==
null
||
direction
==
TraversalDirection
.
UNKNOWN
)
{
if
(
direction
==
null
||
direction
==
TraversalDirection
.
UNKNOWN
)
{
getConnectedEntityGuids
(
entity
,
context
,
TraversalDirection
.
OUTWARD
,
TraversalDirection
.
INWARD
);
getConnectedEntityGuids
(
entity
,
context
,
TraversalDirection
.
OUTWARD
,
TraversalDirection
.
INWARD
);
}
else
{
}
else
{
...
@@ -688,8 +694,7 @@ public class ExportService {
...
@@ -688,8 +694,7 @@ public class ExportService {
private
final
ExportFetchType
fetchType
;
private
final
ExportFetchType
fetchType
;
private
final
String
matchType
;
private
final
String
matchType
;
private
final
boolean
skipLineage
;
private
final
boolean
skipLineage
;
private
final
long
lastModifiedTimestampRequested
;
private
final
long
changeMarker
;
private
long
newestLastModifiedTimestamp
;
private
int
progressReportCount
=
0
;
private
int
progressReportCount
=
0
;
...
@@ -699,45 +704,16 @@ public class ExportService {
...
@@ -699,45 +704,16 @@ public class ExportService {
scriptEngine
=
atlasGraph
.
getGremlinScriptEngine
();
scriptEngine
=
atlasGraph
.
getGremlinScriptEngine
();
bindings
=
new
HashMap
<>();
bindings
=
new
HashMap
<>();
fetchType
=
getFetchType
(
result
.
getRequest
());
fetchType
=
ExportFetchType
.
from
(
result
.
getRequest
().
getFetchTypeOptionValue
());
matchType
=
getMatchType
(
result
.
getRequest
());
matchType
=
result
.
getRequest
().
getMatchTypeOptionValue
();
skipLineage
=
getOptionSkipLineage
(
result
.
getRequest
());
skipLineage
=
result
.
getRequest
().
getSkipLineageOptionValue
();
this
.
lastModifiedTimestampRequested
=
getLastModifiedTimestamp
(
fetchType
,
result
.
getRequest
());
this
.
changeMarker
=
getChangeTokenFromOptions
(
fetchType
,
result
.
getRequest
());
this
.
newestLastModifiedTimestamp
=
0
;
}
private
ExportFetchType
getFetchType
(
AtlasExportRequest
request
)
{
Object
fetchOption
=
request
.
getOptions
()
!=
null
?
request
.
getOptions
().
get
(
OPTION_FETCH_TYPE
)
:
null
;
if
(
fetchOption
instanceof
String
)
{
return
ExportFetchType
.
from
((
String
)
fetchOption
);
}
else
if
(
fetchOption
instanceof
ExportFetchType
)
{
return
(
ExportFetchType
)
fetchOption
;
}
}
return
ExportFetchType
.
FULL
;
private
long
getChangeTokenFromOptions
(
ExportFetchType
fetchType
,
AtlasExportRequest
request
)
{
}
if
(
fetchType
==
ExportFetchType
.
INCREMENTAL
&&
request
.
getOptions
().
containsKey
(
AtlasExportRequest
.
FETCH_TYPE_INCREMENTAL_CHANGE_MARKER
))
{
private
String
getMatchType
(
AtlasExportRequest
request
)
{
return
Long
.
parseLong
(
request
.
getOptions
().
get
(
AtlasExportRequest
.
FETCH_TYPE_INCREMENTAL_CHANGE_MARKER
).
toString
());
String
matchType
=
null
;
if
(
MapUtils
.
isNotEmpty
(
request
.
getOptions
()))
{
if
(
request
.
getOptions
().
get
(
OPTION_ATTR_MATCH_TYPE
)
!=
null
)
{
matchType
=
request
.
getOptions
().
get
(
OPTION_ATTR_MATCH_TYPE
).
toString
();
}
}
return
matchType
;
}
private
boolean
getOptionSkipLineage
(
AtlasExportRequest
request
)
{
return
request
.
getOptions
().
containsKey
(
AtlasExportRequest
.
OPTION_SKIP_LINEAGE
)
&&
(
boolean
)
request
.
getOptions
().
get
(
AtlasExportRequest
.
OPTION_SKIP_LINEAGE
);
}
private
long
getLastModifiedTimestamp
(
ExportFetchType
fetchType
,
AtlasExportRequest
request
)
{
if
(
fetchType
==
ExportFetchType
.
INCREMENTAL
&&
request
.
getOptions
().
containsKey
(
AtlasExportRequest
.
FETCH_TYPE_INCREMENTAL_FROM_TIME
))
{
return
Long
.
parseLong
(
request
.
getOptions
().
get
(
AtlasExportRequest
.
FETCH_TYPE_INCREMENTAL_FROM_TIME
).
toString
());
}
}
return
0L
;
return
0L
;
...
@@ -794,19 +770,7 @@ public class ExportService {
...
@@ -794,19 +770,7 @@ public class ExportService {
return
true
;
return
true
;
}
}
long
entityModificationTimestamp
=
entity
.
getUpdateTime
().
getTime
();
return
changeMarker
<=
entity
.
getUpdateTime
().
getTime
();
updateNewestLastModifiedTimestamp
(
entityModificationTimestamp
);
return
doesTimestampQualify
(
entityModificationTimestamp
);
}
private
void
updateNewestLastModifiedTimestamp
(
long
entityModificationTimestamp
)
{
if
(
newestLastModifiedTimestamp
<
entityModificationTimestamp
)
{
newestLastModifiedTimestamp
=
entityModificationTimestamp
;
}
}
private
boolean
doesTimestampQualify
(
long
modificationTimestamp
)
{
return
lastModifiedTimestampRequested
<
modificationTimestamp
;
}
}
public
boolean
getSkipLineage
()
{
public
boolean
getSkipLineage
()
{
...
@@ -814,7 +778,6 @@ public class ExportService {
...
@@ -814,7 +778,6 @@ public class ExportService {
}
}
public
void
addToSink
(
AtlasEntityWithExtInfo
entityWithExtInfo
)
throws
AtlasBaseException
{
public
void
addToSink
(
AtlasEntityWithExtInfo
entityWithExtInfo
)
throws
AtlasBaseException
{
updateNewestLastModifiedTimestamp
(
entityWithExtInfo
.
getEntity
().
getUpdateTime
().
getTime
());
sink
.
add
(
entityWithExtInfo
);
sink
.
add
(
entityWithExtInfo
);
}
}
}
}
...
...
This diff is collapsed.
Click to expand it.
repository/src/main/java/org/apache/atlas/repository/impexp/ZipSink.java
View file @
8903c9a6
...
@@ -36,10 +36,11 @@ import java.util.zip.ZipOutputStream;
...
@@ -36,10 +36,11 @@ import java.util.zip.ZipOutputStream;
public
class
ZipSink
{
public
class
ZipSink
{
private
static
final
Logger
LOG
=
LoggerFactory
.
getLogger
(
ZipSink
.
class
);
private
static
final
Logger
LOG
=
LoggerFactory
.
getLogger
(
ZipSink
.
class
);
private
static
String
FILE_EXTENSION_JSON
=
".json"
;
private
ZipOutputStream
zipOutputStream
;
private
ZipOutputStream
zipOutputStream
;
final
Set
<
String
>
guids
=
new
HashSet
<>();
final
Set
<
String
>
guids
=
new
HashSet
<>();
public
ZipSink
(
OutputStream
outputStream
)
{
public
ZipSink
(
OutputStream
outputStream
)
{
zipOutputStream
=
new
ZipOutputStream
(
outputStream
);
zipOutputStream
=
new
ZipOutputStream
(
outputStream
);
}
}
...
@@ -92,7 +93,7 @@ public class ZipSink {
...
@@ -92,7 +93,7 @@ public class ZipSink {
private
void
saveToZip
(
String
fileName
,
String
jsonData
)
throws
AtlasBaseException
{
private
void
saveToZip
(
String
fileName
,
String
jsonData
)
throws
AtlasBaseException
{
try
{
try
{
addToZipStream
(
fileName
.
toString
()
+
".json"
,
jsonData
);
addToZipStream
(
fileName
.
toString
()
+
FILE_EXTENSION_JSON
,
jsonData
);
}
catch
(
IOException
e
)
{
}
catch
(
IOException
e
)
{
throw
new
AtlasBaseException
(
String
.
format
(
"Error writing file %s."
,
fileName
),
e
);
throw
new
AtlasBaseException
(
String
.
format
(
"Error writing file %s."
,
fileName
),
e
);
}
}
...
...
This diff is collapsed.
Click to expand it.
repository/src/test/java/org/apache/atlas/repository/impexp/ExportIncrementalTest.java
View file @
8903c9a6
...
@@ -41,7 +41,7 @@ import javax.inject.Inject;
...
@@ -41,7 +41,7 @@ import javax.inject.Inject;
import
java.io.IOException
;
import
java.io.IOException
;
import
java.util.Map
;
import
java.util.Map
;
import
static
org
.
apache
.
atlas
.
model
.
impexp
.
AtlasExportRequest
.
FETCH_TYPE_INCREMENTAL_
FROM_TIME
;
import
static
org
.
apache
.
atlas
.
model
.
impexp
.
AtlasExportRequest
.
FETCH_TYPE_INCREMENTAL_
CHANGE_MARKER
;
import
static
org
.
apache
.
atlas
.
repository
.
impexp
.
ZipFileResourceTestUtils
.
createTypes
;
import
static
org
.
apache
.
atlas
.
repository
.
impexp
.
ZipFileResourceTestUtils
.
createTypes
;
import
static
org
.
apache
.
atlas
.
repository
.
impexp
.
ZipFileResourceTestUtils
.
getEntities
;
import
static
org
.
apache
.
atlas
.
repository
.
impexp
.
ZipFileResourceTestUtils
.
getEntities
;
import
static
org
.
apache
.
atlas
.
repository
.
impexp
.
ZipFileResourceTestUtils
.
runExportWithParameters
;
import
static
org
.
apache
.
atlas
.
repository
.
impexp
.
ZipFileResourceTestUtils
.
runExportWithParameters
;
...
@@ -98,7 +98,7 @@ public class ExportIncrementalTest extends ExportImportTestBase {
...
@@ -98,7 +98,7 @@ public class ExportIncrementalTest extends ExportImportTestBase {
}
}
private
long
updateTimesampForNextIncrementalExport
(
ZipSource
source
)
throws
AtlasBaseException
{
private
long
updateTimesampForNextIncrementalExport
(
ZipSource
source
)
throws
AtlasBaseException
{
return
source
.
getExportResult
().
get
LastModifiedTimestamp
();
return
source
.
getExportResult
().
get
ChangeMarker
();
}
}
@Test
(
dependsOnMethods
=
"atT0_ReturnsAllEntities"
)
@Test
(
dependsOnMethods
=
"atT0_ReturnsAllEntities"
)
...
@@ -161,7 +161,7 @@ public class ExportIncrementalTest extends ExportImportTestBase {
...
@@ -161,7 +161,7 @@ public class ExportIncrementalTest extends ExportImportTestBase {
private
AtlasExportRequest
getIncrementalRequest
(
long
timestamp
)
{
private
AtlasExportRequest
getIncrementalRequest
(
long
timestamp
)
{
try
{
try
{
AtlasExportRequest
request
=
TestResourceFileUtils
.
readObjectFromJson
(
ENTITIES_SUB_DIR
,
EXPORT_REQUEST_INCREMENTAL
,
AtlasExportRequest
.
class
);
AtlasExportRequest
request
=
TestResourceFileUtils
.
readObjectFromJson
(
ENTITIES_SUB_DIR
,
EXPORT_REQUEST_INCREMENTAL
,
AtlasExportRequest
.
class
);
request
.
getOptions
().
put
(
FETCH_TYPE_INCREMENTAL_
FROM_TIME
,
timestamp
);
request
.
getOptions
().
put
(
FETCH_TYPE_INCREMENTAL_
CHANGE_MARKER
,
timestamp
);
return
request
;
return
request
;
}
catch
(
IOException
e
)
{
}
catch
(
IOException
e
)
{
...
...
This diff is collapsed.
Click to expand it.
repository/src/test/java/org/apache/atlas/repository/impexp/ImportTransformsShaperTest.java
View file @
8903c9a6
...
@@ -72,6 +72,7 @@ public class ImportTransformsShaperTest extends ExportImportTestBase {
...
@@ -72,6 +72,7 @@ public class ImportTransformsShaperTest extends ExportImportTestBase {
assertNotNull
(
classification
);
assertNotNull
(
classification
);
assertEntities
(
result
.
getProcessedEntities
(),
TAG_NAME
);
assertEntities
(
result
.
getProcessedEntities
(),
TAG_NAME
);
}
}
private
void
assertEntities
(
List
<
String
>
entityGuids
,
String
tagName
)
throws
AtlasBaseException
{
private
void
assertEntities
(
List
<
String
>
entityGuids
,
String
tagName
)
throws
AtlasBaseException
{
for
(
String
guid
:
entityGuids
)
{
for
(
String
guid
:
entityGuids
)
{
AtlasEntity
.
AtlasEntityWithExtInfo
entityWithExtInfo
=
this
.
entityStore
.
getById
(
guid
);
AtlasEntity
.
AtlasEntityWithExtInfo
entityWithExtInfo
=
this
.
entityStore
.
getById
(
guid
);
...
...
This diff is collapsed.
Click to expand it.
repository/src/test/java/org/apache/atlas/repository/impexp/ReplicationEntityAttributeTest.java
View file @
8903c9a6
...
@@ -159,7 +159,7 @@ public class ReplicationEntityAttributeTest extends ExportImportTestBase {
...
@@ -159,7 +159,7 @@ public class ReplicationEntityAttributeTest extends ExportImportTestBase {
long
actualLastModifiedTimestamp
=
(
long
)
cluster
.
getAdditionalInfoRepl
(
entity
.
getEntity
().
getGuid
());
long
actualLastModifiedTimestamp
=
(
long
)
cluster
.
getAdditionalInfoRepl
(
entity
.
getEntity
().
getGuid
());
assertTrue
(
cluster
.
getAdditionalInfo
().
size
()
>
0
);
assertTrue
(
cluster
.
getAdditionalInfo
().
size
()
>
0
);
assertEquals
(
actualLastModifiedTimestamp
,
importResult
.
getExportResult
().
get
LastModifiedTimestamp
());
assertEquals
(
actualLastModifiedTimestamp
,
importResult
.
getExportResult
().
get
ChangeMarker
());
}
}
private
AtlasExportRequest
getUpdateMetaInfoUpdateRequest
()
{
private
AtlasExportRequest
getUpdateMetaInfoUpdateRequest
()
{
...
...
This diff is collapsed.
Click to expand it.
repository/src/test/java/org/apache/atlas/repository/impexp/ZipSinkTest.java
View file @
8903c9a6
...
@@ -59,7 +59,7 @@ public class ZipSinkTest {
...
@@ -59,7 +59,7 @@ public class ZipSinkTest {
itemsToExport
.
add
(
new
AtlasObjectId
(
"hive_db"
,
"qualifiedName"
,
"default"
));
itemsToExport
.
add
(
new
AtlasObjectId
(
"hive_db"
,
"qualifiedName"
,
"default"
));
request
.
setItemsToExport
(
itemsToExport
);
request
.
setItemsToExport
(
itemsToExport
);
defaultExportResult
=
new
AtlasExportResult
(
request
,
"admin"
,
"1.0.0.0"
,
"root"
,
100
);
defaultExportResult
=
new
AtlasExportResult
(
request
,
"admin"
,
"1.0.0.0"
,
"root"
,
100
,
0L
);
return
defaultExportResult
;
return
defaultExportResult
;
}
}
...
...
This diff is collapsed.
Click to expand it.
repository/src/test/resources/json/stocksDB-Entities/export-incremental.json
View file @
8903c9a6
...
@@ -6,6 +6,6 @@
...
@@ -6,6 +6,6 @@
],
],
"options"
:
{
"options"
:
{
"fetchType"
:
"incremental"
,
"fetchType"
:
"incremental"
,
"
fromTime
"
:
0
"
changeMarker
"
:
0
}
}
}
}
This diff is collapsed.
Click to expand it.
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment