@Test public void testProcessPreviewConverter() throws AtlasException { Mapping m = new Mapping(); m.setMappingType(MappingType.MAP); Field source = new SimpleField(); source.setFieldType(FieldType.STRING); source.setValue("404"); Field target = new SimpleField(); target.setFieldType(FieldType.INTEGER); m.getInputField().add(source); m.getOutputField().add(target); context.processPreview(m); assertEquals(Integer.class, target.getValue().getClass()); assertEquals(404, target.getValue()); }
@Override public Audits processPreview(Mapping mapping) throws AtlasException { DefaultAtlasSession session = new DefaultAtlasSession(this); MappingType mappingType = mapping.getMappingType(); FieldGroup sourceFieldGroup = mapping.getInputFieldGroup(); List<Field> sourceFields = mapping.getInputField(); List<Field> targetFields = mapping.getOutputField(); targetFields.forEach(tf -> tf.setValue(null)); if ((sourceFieldGroup == null && sourceFields.isEmpty()) || targetFields.isEmpty()) { return session.getAudits(); } if (sourceFieldGroup != null) { sourceFields = sourceFieldGroup.getField(); } for (Field sf : sourceFields) { if (sf.getFieldType() == null || sf.getValue() == null) { continue; } if (sf.getValue() instanceof String && ((String)sf.getValue()).isEmpty()) { continue; } if (!restoreSourceFieldType(session, sf)) { return session.getAudits(); } } Field sourceField; Field targetField; if (mappingType == null || mappingType == MappingType.MAP) { if (sourceFieldGroup != null) { List<Field> processed = new LinkedList<>(); for (Field f : sourceFieldGroup.getField()) { processed.add(applyFieldActions(session, f)); } sourceFieldGroup.getField().clear(); sourceFieldGroup.getField().addAll(processed); sourceField = applyFieldActions(session, sourceFieldGroup); } else { sourceField = sourceFields.get(0); sourceField = applyFieldActions(session, sourceField); } session.head().setSourceField(sourceField); sourceFieldGroup = sourceField instanceof FieldGroup ? (FieldGroup) sourceField : null; for (Field f : targetFields) { targetField = f; session.head().setTargetField(targetField); if (sourceFieldGroup != null) { if (sourceFieldGroup.getField().size() == 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The group field '%s:%s' Empty group field is detected, skipping", sourceField.getDocId(), sourceField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } Integer index = targetField.getIndex(); AtlasPath targetPath = new AtlasPath(targetField.getPath()); if (targetPath.hasCollection() && !targetPath.isIndexedCollection()) { if (targetFields.size() > 1) { AtlasUtil.addAudit(session, targetField.getDocId(), "It's not yet supported to have a collection field as a part of multiple target fields in a same mapping", targetField.getPath(), AuditStatus.ERROR, null); return session.getAudits(); } session.head().setSourceField(sourceFieldGroup); } else if (index == null) { session.head().setSourceField(sourceFieldGroup.getField().get(sourceFieldGroup.getField().size()-1)); } else { if (sourceFieldGroup.getField().size() > index) { session.head().setSourceField(sourceFieldGroup.getField().get(index)); } else { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The number of source fields '%s' is fewer than expected via target field index '%s'", sourceFieldGroup.getField().size(), targetField.getIndex()), targetField.getPath(), AuditStatus.WARN, null); continue; } } } if (session.hasErrors()) { return session.getAudits(); } if (!convertSourceToTarget(session, session.head().getSourceField(), targetField)) { return session.getAudits(); } Field processed = applyFieldActions(session, targetField); targetField.setValue(processed.getValue()); } } else if (mappingType == MappingType.COMBINE) { targetField = targetFields.get(0); sourceFields.forEach(sf -> applyFieldActions(session, sf)); Field combined = processCombineField(session, mapping, sourceFields, targetField); if (!convertSourceToTarget(session, combined, targetField)) { return session.getAudits(); } applyFieldActions(session, targetField); } else if (mappingType == MappingType.SEPARATE) { sourceField = sourceFields.get(0); applyFieldActions(session, sourceField); List<Field> separatedFields; try { separatedFields = processSeparateField(session, mapping, sourceField); } catch (AtlasException e) { AtlasUtil.addAudit(session, sourceField.getDocId(), String.format( "Failed to separate field: %s", AtlasUtil.getChainedMessage(e)), sourceField.getPath(), AuditStatus.ERROR, null); if (LOG.isDebugEnabled()) { LOG.error("", e); } return session.getAudits(); } if (separatedFields == null) { return session.getAudits(); } for (Field f : targetFields) { targetField = f; if (targetField.getIndex() == null || targetField.getIndex() < 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "Separate requires zero or positive Index value to be set on targetField targetField.path=%s", targetField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } if (separatedFields.size() <= targetField.getIndex()) { String errorMessage = String.format( "Separate returned fewer segments count=%s when targetField.path=%s requested index=%s", separatedFields.size(), targetField.getPath(), targetField.getIndex()); AtlasUtil.addAudit(session, targetField.getDocId(), errorMessage, targetField.getPath(), AuditStatus.WARN, null); break; } if (!convertSourceToTarget(session, separatedFields.get(targetField.getIndex()), targetField)) { break; } applyFieldActions(session, targetField); } } else { AtlasUtil.addAudit(session, null, String.format( "Unsupported mappingType=%s detected", mapping.getMappingType()), null, AuditStatus.ERROR, null); } return session.getAudits(); }
DefaultAtlasContext implements AtlasContext, AtlasContextMXBean { @Override public Audits processPreview(Mapping mapping) throws AtlasException { DefaultAtlasSession session = new DefaultAtlasSession(this); MappingType mappingType = mapping.getMappingType(); FieldGroup sourceFieldGroup = mapping.getInputFieldGroup(); List<Field> sourceFields = mapping.getInputField(); List<Field> targetFields = mapping.getOutputField(); targetFields.forEach(tf -> tf.setValue(null)); if ((sourceFieldGroup == null && sourceFields.isEmpty()) || targetFields.isEmpty()) { return session.getAudits(); } if (sourceFieldGroup != null) { sourceFields = sourceFieldGroup.getField(); } for (Field sf : sourceFields) { if (sf.getFieldType() == null || sf.getValue() == null) { continue; } if (sf.getValue() instanceof String && ((String)sf.getValue()).isEmpty()) { continue; } if (!restoreSourceFieldType(session, sf)) { return session.getAudits(); } } Field sourceField; Field targetField; if (mappingType == null || mappingType == MappingType.MAP) { if (sourceFieldGroup != null) { List<Field> processed = new LinkedList<>(); for (Field f : sourceFieldGroup.getField()) { processed.add(applyFieldActions(session, f)); } sourceFieldGroup.getField().clear(); sourceFieldGroup.getField().addAll(processed); sourceField = applyFieldActions(session, sourceFieldGroup); } else { sourceField = sourceFields.get(0); sourceField = applyFieldActions(session, sourceField); } session.head().setSourceField(sourceField); sourceFieldGroup = sourceField instanceof FieldGroup ? (FieldGroup) sourceField : null; for (Field f : targetFields) { targetField = f; session.head().setTargetField(targetField); if (sourceFieldGroup != null) { if (sourceFieldGroup.getField().size() == 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The group field '%s:%s' Empty group field is detected, skipping", sourceField.getDocId(), sourceField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } Integer index = targetField.getIndex(); AtlasPath targetPath = new AtlasPath(targetField.getPath()); if (targetPath.hasCollection() && !targetPath.isIndexedCollection()) { if (targetFields.size() > 1) { AtlasUtil.addAudit(session, targetField.getDocId(), "It's not yet supported to have a collection field as a part of multiple target fields in a same mapping", targetField.getPath(), AuditStatus.ERROR, null); return session.getAudits(); } session.head().setSourceField(sourceFieldGroup); } else if (index == null) { session.head().setSourceField(sourceFieldGroup.getField().get(sourceFieldGroup.getField().size()-1)); } else { if (sourceFieldGroup.getField().size() > index) { session.head().setSourceField(sourceFieldGroup.getField().get(index)); } else { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The number of source fields '%s' is fewer than expected via target field index '%s'", sourceFieldGroup.getField().size(), targetField.getIndex()), targetField.getPath(), AuditStatus.WARN, null); continue; } } } if (session.hasErrors()) { return session.getAudits(); } if (!convertSourceToTarget(session, session.head().getSourceField(), targetField)) { return session.getAudits(); } Field processed = applyFieldActions(session, targetField); targetField.setValue(processed.getValue()); } } else if (mappingType == MappingType.COMBINE) { targetField = targetFields.get(0); sourceFields.forEach(sf -> applyFieldActions(session, sf)); Field combined = processCombineField(session, mapping, sourceFields, targetField); if (!convertSourceToTarget(session, combined, targetField)) { return session.getAudits(); } applyFieldActions(session, targetField); } else if (mappingType == MappingType.SEPARATE) { sourceField = sourceFields.get(0); applyFieldActions(session, sourceField); List<Field> separatedFields; try { separatedFields = processSeparateField(session, mapping, sourceField); } catch (AtlasException e) { AtlasUtil.addAudit(session, sourceField.getDocId(), String.format( "Failed to separate field: %s", AtlasUtil.getChainedMessage(e)), sourceField.getPath(), AuditStatus.ERROR, null); if (LOG.isDebugEnabled()) { LOG.error("", e); } return session.getAudits(); } if (separatedFields == null) { return session.getAudits(); } for (Field f : targetFields) { targetField = f; if (targetField.getIndex() == null || targetField.getIndex() < 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "Separate requires zero or positive Index value to be set on targetField targetField.path=%s", targetField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } if (separatedFields.size() <= targetField.getIndex()) { String errorMessage = String.format( "Separate returned fewer segments count=%s when targetField.path=%s requested index=%s", separatedFields.size(), targetField.getPath(), targetField.getIndex()); AtlasUtil.addAudit(session, targetField.getDocId(), errorMessage, targetField.getPath(), AuditStatus.WARN, null); break; } if (!convertSourceToTarget(session, separatedFields.get(targetField.getIndex()), targetField)) { break; } applyFieldActions(session, targetField); } } else { AtlasUtil.addAudit(session, null, String.format( "Unsupported mappingType=%s detected", mapping.getMappingType()), null, AuditStatus.ERROR, null); } return session.getAudits(); } }
DefaultAtlasContext implements AtlasContext, AtlasContextMXBean { @Override public Audits processPreview(Mapping mapping) throws AtlasException { DefaultAtlasSession session = new DefaultAtlasSession(this); MappingType mappingType = mapping.getMappingType(); FieldGroup sourceFieldGroup = mapping.getInputFieldGroup(); List<Field> sourceFields = mapping.getInputField(); List<Field> targetFields = mapping.getOutputField(); targetFields.forEach(tf -> tf.setValue(null)); if ((sourceFieldGroup == null && sourceFields.isEmpty()) || targetFields.isEmpty()) { return session.getAudits(); } if (sourceFieldGroup != null) { sourceFields = sourceFieldGroup.getField(); } for (Field sf : sourceFields) { if (sf.getFieldType() == null || sf.getValue() == null) { continue; } if (sf.getValue() instanceof String && ((String)sf.getValue()).isEmpty()) { continue; } if (!restoreSourceFieldType(session, sf)) { return session.getAudits(); } } Field sourceField; Field targetField; if (mappingType == null || mappingType == MappingType.MAP) { if (sourceFieldGroup != null) { List<Field> processed = new LinkedList<>(); for (Field f : sourceFieldGroup.getField()) { processed.add(applyFieldActions(session, f)); } sourceFieldGroup.getField().clear(); sourceFieldGroup.getField().addAll(processed); sourceField = applyFieldActions(session, sourceFieldGroup); } else { sourceField = sourceFields.get(0); sourceField = applyFieldActions(session, sourceField); } session.head().setSourceField(sourceField); sourceFieldGroup = sourceField instanceof FieldGroup ? (FieldGroup) sourceField : null; for (Field f : targetFields) { targetField = f; session.head().setTargetField(targetField); if (sourceFieldGroup != null) { if (sourceFieldGroup.getField().size() == 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The group field '%s:%s' Empty group field is detected, skipping", sourceField.getDocId(), sourceField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } Integer index = targetField.getIndex(); AtlasPath targetPath = new AtlasPath(targetField.getPath()); if (targetPath.hasCollection() && !targetPath.isIndexedCollection()) { if (targetFields.size() > 1) { AtlasUtil.addAudit(session, targetField.getDocId(), "It's not yet supported to have a collection field as a part of multiple target fields in a same mapping", targetField.getPath(), AuditStatus.ERROR, null); return session.getAudits(); } session.head().setSourceField(sourceFieldGroup); } else if (index == null) { session.head().setSourceField(sourceFieldGroup.getField().get(sourceFieldGroup.getField().size()-1)); } else { if (sourceFieldGroup.getField().size() > index) { session.head().setSourceField(sourceFieldGroup.getField().get(index)); } else { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The number of source fields '%s' is fewer than expected via target field index '%s'", sourceFieldGroup.getField().size(), targetField.getIndex()), targetField.getPath(), AuditStatus.WARN, null); continue; } } } if (session.hasErrors()) { return session.getAudits(); } if (!convertSourceToTarget(session, session.head().getSourceField(), targetField)) { return session.getAudits(); } Field processed = applyFieldActions(session, targetField); targetField.setValue(processed.getValue()); } } else if (mappingType == MappingType.COMBINE) { targetField = targetFields.get(0); sourceFields.forEach(sf -> applyFieldActions(session, sf)); Field combined = processCombineField(session, mapping, sourceFields, targetField); if (!convertSourceToTarget(session, combined, targetField)) { return session.getAudits(); } applyFieldActions(session, targetField); } else if (mappingType == MappingType.SEPARATE) { sourceField = sourceFields.get(0); applyFieldActions(session, sourceField); List<Field> separatedFields; try { separatedFields = processSeparateField(session, mapping, sourceField); } catch (AtlasException e) { AtlasUtil.addAudit(session, sourceField.getDocId(), String.format( "Failed to separate field: %s", AtlasUtil.getChainedMessage(e)), sourceField.getPath(), AuditStatus.ERROR, null); if (LOG.isDebugEnabled()) { LOG.error("", e); } return session.getAudits(); } if (separatedFields == null) { return session.getAudits(); } for (Field f : targetFields) { targetField = f; if (targetField.getIndex() == null || targetField.getIndex() < 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "Separate requires zero or positive Index value to be set on targetField targetField.path=%s", targetField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } if (separatedFields.size() <= targetField.getIndex()) { String errorMessage = String.format( "Separate returned fewer segments count=%s when targetField.path=%s requested index=%s", separatedFields.size(), targetField.getPath(), targetField.getIndex()); AtlasUtil.addAudit(session, targetField.getDocId(), errorMessage, targetField.getPath(), AuditStatus.WARN, null); break; } if (!convertSourceToTarget(session, separatedFields.get(targetField.getIndex()), targetField)) { break; } applyFieldActions(session, targetField); } } else { AtlasUtil.addAudit(session, null, String.format( "Unsupported mappingType=%s detected", mapping.getMappingType()), null, AuditStatus.ERROR, null); } return session.getAudits(); } DefaultAtlasContext(URI atlasMappingUri);  DefaultAtlasContext(DefaultAtlasContextFactory factory, URI atlasMappingUri);  DefaultAtlasContext(DefaultAtlasContextFactory factory, AtlasMapping mapping);  DefaultAtlasContext(DefaultAtlasContextFactory factory,
            AtlasContextFactory.Format format, InputStream stream); }
DefaultAtlasContext implements AtlasContext, AtlasContextMXBean { @Override public Audits processPreview(Mapping mapping) throws AtlasException { DefaultAtlasSession session = new DefaultAtlasSession(this); MappingType mappingType = mapping.getMappingType(); FieldGroup sourceFieldGroup = mapping.getInputFieldGroup(); List<Field> sourceFields = mapping.getInputField(); List<Field> targetFields = mapping.getOutputField(); targetFields.forEach(tf -> tf.setValue(null)); if ((sourceFieldGroup == null && sourceFields.isEmpty()) || targetFields.isEmpty()) { return session.getAudits(); } if (sourceFieldGroup != null) { sourceFields = sourceFieldGroup.getField(); } for (Field sf : sourceFields) { if (sf.getFieldType() == null || sf.getValue() == null) { continue; } if (sf.getValue() instanceof String && ((String)sf.getValue()).isEmpty()) { continue; } if (!restoreSourceFieldType(session, sf)) { return session.getAudits(); } } Field sourceField; Field targetField; if (mappingType == null || mappingType == MappingType.MAP) { if (sourceFieldGroup != null) { List<Field> processed = new LinkedList<>(); for (Field f : sourceFieldGroup.getField()) { processed.add(applyFieldActions(session, f)); } sourceFieldGroup.getField().clear(); sourceFieldGroup.getField().addAll(processed); sourceField = applyFieldActions(session, sourceFieldGroup); } else { sourceField = sourceFields.get(0); sourceField = applyFieldActions(session, sourceField); } session.head().setSourceField(sourceField); sourceFieldGroup = sourceField instanceof FieldGroup ? (FieldGroup) sourceField : null; for (Field f : targetFields) { targetField = f; session.head().setTargetField(targetField); if (sourceFieldGroup != null) { if (sourceFieldGroup.getField().size() == 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The group field '%s:%s' Empty group field is detected, skipping", sourceField.getDocId(), sourceField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } Integer index = targetField.getIndex(); AtlasPath targetPath = new AtlasPath(targetField.getPath()); if (targetPath.hasCollection() && !targetPath.isIndexedCollection()) { if (targetFields.size() > 1) { AtlasUtil.addAudit(session, targetField.getDocId(), "It's not yet supported to have a collection field as a part of multiple target fields in a same mapping", targetField.getPath(), AuditStatus.ERROR, null); return session.getAudits(); } session.head().setSourceField(sourceFieldGroup); } else if (index == null) { session.head().setSourceField(sourceFieldGroup.getField().get(sourceFieldGroup.getField().size()-1)); } else { if (sourceFieldGroup.getField().size() > index) { session.head().setSourceField(sourceFieldGroup.getField().get(index)); } else { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The number of source fields '%s' is fewer than expected via target field index '%s'", sourceFieldGroup.getField().size(), targetField.getIndex()), targetField.getPath(), AuditStatus.WARN, null); continue; } } } if (session.hasErrors()) { return session.getAudits(); } if (!convertSourceToTarget(session, session.head().getSourceField(), targetField)) { return session.getAudits(); } Field processed = applyFieldActions(session, targetField); targetField.setValue(processed.getValue()); } } else if (mappingType == MappingType.COMBINE) { targetField = targetFields.get(0); sourceFields.forEach(sf -> applyFieldActions(session, sf)); Field combined = processCombineField(session, mapping, sourceFields, targetField); if (!convertSourceToTarget(session, combined, targetField)) { return session.getAudits(); } applyFieldActions(session, targetField); } else if (mappingType == MappingType.SEPARATE) { sourceField = sourceFields.get(0); applyFieldActions(session, sourceField); List<Field> separatedFields; try { separatedFields = processSeparateField(session, mapping, sourceField); } catch (AtlasException e) { AtlasUtil.addAudit(session, sourceField.getDocId(), String.format( "Failed to separate field: %s", AtlasUtil.getChainedMessage(e)), sourceField.getPath(), AuditStatus.ERROR, null); if (LOG.isDebugEnabled()) { LOG.error("", e); } return session.getAudits(); } if (separatedFields == null) { return session.getAudits(); } for (Field f : targetFields) { targetField = f; if (targetField.getIndex() == null || targetField.getIndex() < 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "Separate requires zero or positive Index value to be set on targetField targetField.path=%s", targetField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } if (separatedFields.size() <= targetField.getIndex()) { String errorMessage = String.format( "Separate returned fewer segments count=%s when targetField.path=%s requested index=%s", separatedFields.size(), targetField.getPath(), targetField.getIndex()); AtlasUtil.addAudit(session, targetField.getDocId(), errorMessage, targetField.getPath(), AuditStatus.WARN, null); break; } if (!convertSourceToTarget(session, separatedFields.get(targetField.getIndex()), targetField)) { break; } applyFieldActions(session, targetField); } } else { AtlasUtil.addAudit(session, null, String.format( "Unsupported mappingType=%s detected", mapping.getMappingType()), null, AuditStatus.ERROR, null); } return session.getAudits(); } DefaultAtlasContext(URI atlasMappingUri);  DefaultAtlasContext(DefaultAtlasContextFactory factory, URI atlasMappingUri);  DefaultAtlasContext(DefaultAtlasContextFactory factory, AtlasMapping mapping);  DefaultAtlasContext(DefaultAtlasContextFactory factory,
            AtlasContextFactory.Format format, InputStream stream); @Override Audits processPreview(Mapping mapping); @Override void process(AtlasSession userSession); @Override void processValidation(AtlasSession userSession); @Override AtlasContextFactory getContextFactory(); AtlasMapping getMapping(); @Override AtlasSession createSession(); AtlasSession createSession(AtlasMapping mappingDefinition); Map<String, AtlasModule> getSourceModules(); void setSourceModules(Map<String, AtlasModule> sourceModules); Map<String, AtlasModule> getTargetModules(); void setTargetModules(Map<String, AtlasModule> targetModules); Map<String, LookupTable> getLookupTables(); void setLookupTables(Map<String, LookupTable> lookupTables); ObjectName getJmxObjectName(); @Override String getUuid(); @Override String getVersion(); @Override String getMappingName(); @Override String getMappingUri(); @Override String getClassName(); @Override String getThreadName(); @Override String toString(); ADMArchiveHandler getADMArchiveHandler(); }
DefaultAtlasContext implements AtlasContext, AtlasContextMXBean { @Override public Audits processPreview(Mapping mapping) throws AtlasException { DefaultAtlasSession session = new DefaultAtlasSession(this); MappingType mappingType = mapping.getMappingType(); FieldGroup sourceFieldGroup = mapping.getInputFieldGroup(); List<Field> sourceFields = mapping.getInputField(); List<Field> targetFields = mapping.getOutputField(); targetFields.forEach(tf -> tf.setValue(null)); if ((sourceFieldGroup == null && sourceFields.isEmpty()) || targetFields.isEmpty()) { return session.getAudits(); } if (sourceFieldGroup != null) { sourceFields = sourceFieldGroup.getField(); } for (Field sf : sourceFields) { if (sf.getFieldType() == null || sf.getValue() == null) { continue; } if (sf.getValue() instanceof String && ((String)sf.getValue()).isEmpty()) { continue; } if (!restoreSourceFieldType(session, sf)) { return session.getAudits(); } } Field sourceField; Field targetField; if (mappingType == null || mappingType == MappingType.MAP) { if (sourceFieldGroup != null) { List<Field> processed = new LinkedList<>(); for (Field f : sourceFieldGroup.getField()) { processed.add(applyFieldActions(session, f)); } sourceFieldGroup.getField().clear(); sourceFieldGroup.getField().addAll(processed); sourceField = applyFieldActions(session, sourceFieldGroup); } else { sourceField = sourceFields.get(0); sourceField = applyFieldActions(session, sourceField); } session.head().setSourceField(sourceField); sourceFieldGroup = sourceField instanceof FieldGroup ? (FieldGroup) sourceField : null; for (Field f : targetFields) { targetField = f; session.head().setTargetField(targetField); if (sourceFieldGroup != null) { if (sourceFieldGroup.getField().size() == 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The group field '%s:%s' Empty group field is detected, skipping", sourceField.getDocId(), sourceField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } Integer index = targetField.getIndex(); AtlasPath targetPath = new AtlasPath(targetField.getPath()); if (targetPath.hasCollection() && !targetPath.isIndexedCollection()) { if (targetFields.size() > 1) { AtlasUtil.addAudit(session, targetField.getDocId(), "It's not yet supported to have a collection field as a part of multiple target fields in a same mapping", targetField.getPath(), AuditStatus.ERROR, null); return session.getAudits(); } session.head().setSourceField(sourceFieldGroup); } else if (index == null) { session.head().setSourceField(sourceFieldGroup.getField().get(sourceFieldGroup.getField().size()-1)); } else { if (sourceFieldGroup.getField().size() > index) { session.head().setSourceField(sourceFieldGroup.getField().get(index)); } else { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "The number of source fields '%s' is fewer than expected via target field index '%s'", sourceFieldGroup.getField().size(), targetField.getIndex()), targetField.getPath(), AuditStatus.WARN, null); continue; } } } if (session.hasErrors()) { return session.getAudits(); } if (!convertSourceToTarget(session, session.head().getSourceField(), targetField)) { return session.getAudits(); } Field processed = applyFieldActions(session, targetField); targetField.setValue(processed.getValue()); } } else if (mappingType == MappingType.COMBINE) { targetField = targetFields.get(0); sourceFields.forEach(sf -> applyFieldActions(session, sf)); Field combined = processCombineField(session, mapping, sourceFields, targetField); if (!convertSourceToTarget(session, combined, targetField)) { return session.getAudits(); } applyFieldActions(session, targetField); } else if (mappingType == MappingType.SEPARATE) { sourceField = sourceFields.get(0); applyFieldActions(session, sourceField); List<Field> separatedFields; try { separatedFields = processSeparateField(session, mapping, sourceField); } catch (AtlasException e) { AtlasUtil.addAudit(session, sourceField.getDocId(), String.format( "Failed to separate field: %s", AtlasUtil.getChainedMessage(e)), sourceField.getPath(), AuditStatus.ERROR, null); if (LOG.isDebugEnabled()) { LOG.error("", e); } return session.getAudits(); } if (separatedFields == null) { return session.getAudits(); } for (Field f : targetFields) { targetField = f; if (targetField.getIndex() == null || targetField.getIndex() < 0) { AtlasUtil.addAudit(session, targetField.getDocId(), String.format( "Separate requires zero or positive Index value to be set on targetField targetField.path=%s", targetField.getPath()), targetField.getPath(), AuditStatus.WARN, null); continue; } if (separatedFields.size() <= targetField.getIndex()) { String errorMessage = String.format( "Separate returned fewer segments count=%s when targetField.path=%s requested index=%s", separatedFields.size(), targetField.getPath(), targetField.getIndex()); AtlasUtil.addAudit(session, targetField.getDocId(), errorMessage, targetField.getPath(), AuditStatus.WARN, null); break; } if (!convertSourceToTarget(session, separatedFields.get(targetField.getIndex()), targetField)) { break; } applyFieldActions(session, targetField); } } else { AtlasUtil.addAudit(session, null, String.format( "Unsupported mappingType=%s detected", mapping.getMappingType()), null, AuditStatus.ERROR, null); } return session.getAudits(); } DefaultAtlasContext(URI atlasMappingUri);  DefaultAtlasContext(DefaultAtlasContextFactory factory, URI atlasMappingUri);  DefaultAtlasContext(DefaultAtlasContextFactory factory, AtlasMapping mapping);  DefaultAtlasContext(DefaultAtlasContextFactory factory,
            AtlasContextFactory.Format format, InputStream stream); @Override Audits processPreview(Mapping mapping); @Override void process(AtlasSession userSession); @Override void processValidation(AtlasSession userSession); @Override AtlasContextFactory getContextFactory(); AtlasMapping getMapping(); @Override AtlasSession createSession(); AtlasSession createSession(AtlasMapping mappingDefinition); Map<String, AtlasModule> getSourceModules(); void setSourceModules(Map<String, AtlasModule> sourceModules); Map<String, AtlasModule> getTargetModules(); void setTargetModules(Map<String, AtlasModule> targetModules); Map<String, LookupTable> getLookupTables(); void setLookupTables(Map<String, LookupTable> lookupTables); ObjectName getJmxObjectName(); @Override String getUuid(); @Override String getVersion(); @Override String getMappingName(); @Override String getMappingUri(); @Override String getClassName(); @Override String getThreadName(); @Override String toString(); ADMArchiveHandler getADMArchiveHandler(); static final String CONSTANTS_DOCUMENT_ID; static final String PROPERTIES_DOCUMENT_ID; }