@Test public void testNotStructPartition() throws PartitionException { String fieldName = "foo"; FieldPartitioner<String> partitioner = getFieldPartitioner(fieldName); SinkRecord sinkRecord = new SinkRecord(TOPIC, PARTITION, Schema.STRING_SCHEMA, null, Schema.STRING_SCHEMA, fieldName, 0L); Exception e = assertThrows(PartitionException.class, () -> { partitioner.encodePartition(sinkRecord); }); assertEquals("Error encoding partition.", e.getMessage()); }
@Override public String encodePartition(SinkRecord sinkRecord) { Object value = sinkRecord.value(); if (value instanceof Struct) { final Schema valueSchema = sinkRecord.valueSchema(); final Struct struct = (Struct) value; StringBuilder builder = new StringBuilder(); for (String fieldName : fieldNames) { if (builder.length() > 0) { builder.append(this.delim); } Object partitionKey = struct.get(fieldName); Type type = valueSchema.field(fieldName).schema().type(); switch (type) { case INT8: case INT16: case INT32: case INT64: Number record = (Number) partitionKey; builder.append(fieldName + "=" + record.toString()); break; case STRING: builder.append(fieldName + "=" + (String) partitionKey); break; case BOOLEAN: boolean booleanRecord = (boolean) partitionKey; builder.append(fieldName + "=" + Boolean.toString(booleanRecord)); break; default: log.error("Type {} is not supported as a partition key.", type.getName()); throw new PartitionException("Error encoding partition."); } } return builder.toString(); } else { log.error("Value is not Struct type."); throw new PartitionException("Error encoding partition."); } }
FieldPartitioner extends DefaultPartitioner<T> { @Override public String encodePartition(SinkRecord sinkRecord) { Object value = sinkRecord.value(); if (value instanceof Struct) { final Schema valueSchema = sinkRecord.valueSchema(); final Struct struct = (Struct) value; StringBuilder builder = new StringBuilder(); for (String fieldName : fieldNames) { if (builder.length() > 0) { builder.append(this.delim); } Object partitionKey = struct.get(fieldName); Type type = valueSchema.field(fieldName).schema().type(); switch (type) { case INT8: case INT16: case INT32: case INT64: Number record = (Number) partitionKey; builder.append(fieldName + "=" + record.toString()); break; case STRING: builder.append(fieldName + "=" + (String) partitionKey); break; case BOOLEAN: boolean booleanRecord = (boolean) partitionKey; builder.append(fieldName + "=" + Boolean.toString(booleanRecord)); break; default: log.error("Type {} is not supported as a partition key.", type.getName()); throw new PartitionException("Error encoding partition."); } } return builder.toString(); } else { log.error("Value is not Struct type."); throw new PartitionException("Error encoding partition."); } } }
FieldPartitioner extends DefaultPartitioner<T> { @Override public String encodePartition(SinkRecord sinkRecord) { Object value = sinkRecord.value(); if (value instanceof Struct) { final Schema valueSchema = sinkRecord.valueSchema(); final Struct struct = (Struct) value; StringBuilder builder = new StringBuilder(); for (String fieldName : fieldNames) { if (builder.length() > 0) { builder.append(this.delim); } Object partitionKey = struct.get(fieldName); Type type = valueSchema.field(fieldName).schema().type(); switch (type) { case INT8: case INT16: case INT32: case INT64: Number record = (Number) partitionKey; builder.append(fieldName + "=" + record.toString()); break; case STRING: builder.append(fieldName + "=" + (String) partitionKey); break; case BOOLEAN: boolean booleanRecord = (boolean) partitionKey; builder.append(fieldName + "=" + Boolean.toString(booleanRecord)); break; default: log.error("Type {} is not supported as a partition key.", type.getName()); throw new PartitionException("Error encoding partition."); } } return builder.toString(); } else { log.error("Value is not Struct type."); throw new PartitionException("Error encoding partition."); } }  }
FieldPartitioner extends DefaultPartitioner<T> { @Override public String encodePartition(SinkRecord sinkRecord) { Object value = sinkRecord.value(); if (value instanceof Struct) { final Schema valueSchema = sinkRecord.valueSchema(); final Struct struct = (Struct) value; StringBuilder builder = new StringBuilder(); for (String fieldName : fieldNames) { if (builder.length() > 0) { builder.append(this.delim); } Object partitionKey = struct.get(fieldName); Type type = valueSchema.field(fieldName).schema().type(); switch (type) { case INT8: case INT16: case INT32: case INT64: Number record = (Number) partitionKey; builder.append(fieldName + "=" + record.toString()); break; case STRING: builder.append(fieldName + "=" + (String) partitionKey); break; case BOOLEAN: boolean booleanRecord = (boolean) partitionKey; builder.append(fieldName + "=" + Boolean.toString(booleanRecord)); break; default: log.error("Type {} is not supported as a partition key.", type.getName()); throw new PartitionException("Error encoding partition."); } } return builder.toString(); } else { log.error("Value is not Struct type."); throw new PartitionException("Error encoding partition."); } }  @SuppressWarnings("unchecked") @Override void configure(Map<String, Object> config); @Override String encodePartition(SinkRecord sinkRecord); @Override List<T> partitionFields(); }
FieldPartitioner extends DefaultPartitioner<T> { @Override public String encodePartition(SinkRecord sinkRecord) { Object value = sinkRecord.value(); if (value instanceof Struct) { final Schema valueSchema = sinkRecord.valueSchema(); final Struct struct = (Struct) value; StringBuilder builder = new StringBuilder(); for (String fieldName : fieldNames) { if (builder.length() > 0) { builder.append(this.delim); } Object partitionKey = struct.get(fieldName); Type type = valueSchema.field(fieldName).schema().type(); switch (type) { case INT8: case INT16: case INT32: case INT64: Number record = (Number) partitionKey; builder.append(fieldName + "=" + record.toString()); break; case STRING: builder.append(fieldName + "=" + (String) partitionKey); break; case BOOLEAN: boolean booleanRecord = (boolean) partitionKey; builder.append(fieldName + "=" + Boolean.toString(booleanRecord)); break; default: log.error("Type {} is not supported as a partition key.", type.getName()); throw new PartitionException("Error encoding partition."); } } return builder.toString(); } else { log.error("Value is not Struct type."); throw new PartitionException("Error encoding partition."); } }  @SuppressWarnings("unchecked") @Override void configure(Map<String, Object> config); @Override String encodePartition(SinkRecord sinkRecord); @Override List<T> partitionFields();  }