@Override
  public void encode(DataSchema schema) throws IOException
  {
    // Initialize a new builder for the preferred encoding style
    _builder = _encodingStyle.newBuilderInstance(_writer);

    // Set and write root namespace/package
    if (schema instanceof NamedDataSchema)
    {
      NamedDataSchema namedSchema = (NamedDataSchema) schema;
      boolean hasNamespace = StringUtils.isNotBlank(namedSchema.getNamespace());
      boolean hasPackage = StringUtils.isNotBlank(namedSchema.getPackage());
      if (hasNamespace || hasPackage)
      {
        if (hasNamespace)
        {
          markSchemaElementStartLocation();
          _builder.write("namespace")
              .writeSpace()
              .writeIdentifier(namedSchema.getNamespace())
              .newline();
          recordSchemaElementLocation(namedSchema.getNamespace());
          _namespace = namedSchema.getNamespace();
        }
        if (hasPackage)
        {
          _builder.write("package")
              .writeSpace()
              .writeIdentifier(namedSchema.getPackage())
              .newline();
          _package = namedSchema.getPackage();
        }
        _builder.newline();
      }
    }

    // Compute imports
    if (_typeReferenceFormat != TypeReferenceFormat.DENORMALIZE)
    {
      _importsByLocalName = computeImports(schema, _namespace);
    }
    else
    {
      _importsByLocalName = Collections.emptyMap();
    }

    // Write imports sorted by fully qualified name
    if (_importsByLocalName.size() > 0)
    {
      for (Name importName : new TreeSet<>(_importsByLocalName.values()))
      {
        _builder.write("import")
            .writeSpace()
            .writeIdentifier(importName.getFullName())
            .newline();
      }
      _builder.newline();
    }

    // Write the schema
    writeInlineSchema(schema);
  }
@Test
  public void testEncodeRecordWithEmptyDataMapInProperty() throws IOException
  {
    RecordDataSchema source =
        new RecordDataSchema(new Name("com.linkedin.test.RecordDataSchema"), RecordDataSchema.RecordType.RECORD);
    Map<String, Object> properties = new HashMap<>();
    properties.put("empty", new DataMap());
    source.setProperties(properties);

    // schema to pdl
    StringWriter writer = new StringWriter();
    SchemaToPdlEncoder encoder = new SchemaToPdlEncoder(writer);
    encoder.setTypeReferenceFormat(SchemaToPdlEncoder.TypeReferenceFormat.PRESERVE);
    encoder.encode(source);

    DataSchema encoded = TestUtil.dataSchemaFromPdlString(writer.toString());
    assertTrue(encoded instanceof RecordDataSchema);
    assertEquals(source.getProperties(), encoded.getProperties());
    assertEquals(source, encoded);
  }