@Override
  public byte[] fromConnectData(final String topic, final Schema schema, final Object value) {
    if (this.schema == null) {
      throw new UnsupportedOperationException("ProtobufNoSRConverter is an internal "
          + "converter to ksqldb. It should not be instantiated via reflection through a no-arg "
          + "constructor.");
    }
    try {
      final ProtobufSchemaAndValue schemaAndValue = protobufData.fromConnectData(schema, value);
      final Object v = schemaAndValue.getValue();
      if (v == null) {
        return null;
      } else if (v instanceof Message) {
        return serializer.serialize((Message) v);
      } else {
        throw new DataException("Unsupported object of class " + v.getClass().getName());
      }
    } catch (SerializationException e) {
      throw new DataException(String.format(
          "Failed to serialize Protobuf data from topic %s :",
          topic
      ), e);
    } catch (InvalidConfigurationException e) {
      throw new ConfigException(
          String.format("Failed to access Protobuf data from topic %s : %s", topic, e.getMessage())
      );
    }
  }
@Test(expected = UnsupportedOperationException.class)
  public void shouldThrowExceptionWhenUsedWithNoArgConstructor2() {
    // Given
    final ProtobufNoSRConverter protobufNoSRConverter = new ProtobufNoSRConverter();

    // When
    protobufNoSRConverter.fromConnectData("topic", Schema.STRING_SCHEMA, "test");
  }