@Test public void withSchema() { Schema schema = SchemaBuilder.struct() .field("magic", Schema.INT32_SCHEMA) .field("bool", Schema.BOOLEAN_SCHEMA) .field("byte", Schema.INT8_SCHEMA) .field("short", Schema.INT16_SCHEMA) .field("int", Schema.INT32_SCHEMA) .field("long", Schema.INT64_SCHEMA) .field("float", Schema.FLOAT32_SCHEMA) .field("double", Schema.FLOAT64_SCHEMA) .field("string", Schema.STRING_SCHEMA) .field("date", org.apache.kafka.connect.data.Date.SCHEMA) .field("time", Time.SCHEMA) .field("timestamp", Timestamp.SCHEMA) .field("decimal", Decimal.schema(0)) .field("array", SchemaBuilder.array(Schema.INT32_SCHEMA)) .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA)) .build(); final Struct value = new Struct(schema); value.put("magic", 42); value.put("bool", true); value.put("byte", (byte) 42); value.put("short", (short) 42); value.put("int", 42); value.put("long", 42L); value.put("float", 42f); value.put("double", 42d); value.put("string", "hmm"); value.put("date", new Date()); value.put("time", new Date()); value.put("timestamp", new Date()); value.put("decimal", new BigDecimal(42)); value.put("array", Arrays.asList(1, 2, 3)); value.put("map", Collections.singletonMap("what", "what")); final List<String> maskFields = new ArrayList<>(schema.fields().size()); for (Field field: schema.fields()) { if (!field.name().equals("magic")) { maskFields.add(field.name()); } } final Struct updatedValue = (Struct) transform(maskFields).apply(record(schema, value)).value(); assertEquals(42, updatedValue.get("magic")); assertEquals(false, updatedValue.get("bool")); assertEquals((byte) 0, updatedValue.get("byte")); assertEquals((short) 0, updatedValue.get("short")); assertEquals(0, updatedValue.get("int")); assertEquals(0L, updatedValue.get("long")); assertEquals(0f, updatedValue.get("float")); assertEquals(0d, updatedValue.get("double")); assertEquals("", updatedValue.get("string")); assertEquals(new Date(0), updatedValue.get("date")); assertEquals(new Date(0), updatedValue.get("time")); assertEquals(new Date(0), updatedValue.get("timestamp")); assertEquals(BigDecimal.ZERO, updatedValue.get("decimal")); assertEquals(Collections.emptyList(), updatedValue.get("array")); assertEquals(Collections.emptyMap(), updatedValue.get("map")); }
@Override public R apply(R record) { if (operatingSchema(record) == null) { return applySchemaless(record); } else { return applyWithSchema(record); } }
MaskField implements Transformation<R> { @Override public R apply(R record) { if (operatingSchema(record) == null) { return applySchemaless(record); } else { return applyWithSchema(record); } } }
MaskField implements Transformation<R> { @Override public R apply(R record) { if (operatingSchema(record) == null) { return applySchemaless(record); } else { return applyWithSchema(record); } }  }
MaskField implements Transformation<R> { @Override public R apply(R record) { if (operatingSchema(record) == null) { return applySchemaless(record); } else { return applyWithSchema(record); } }  @Override void configure(Map<String, ?> props); @Override R apply(R record); @Override ConfigDef config(); @Override void close(); }
MaskField implements Transformation<R> { @Override public R apply(R record) { if (operatingSchema(record) == null) { return applySchemaless(record); } else { return applyWithSchema(record); } }  @Override void configure(Map<String, ?> props); @Override R apply(R record); @Override ConfigDef config(); @Override void close(); static final String OVERVIEW_DOC; static final ConfigDef CONFIG_DEF; }