@Override
  public SchemaAndValue toConnectData(final String topic, final byte[] 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 ProtobufSchema protobufSchema = protobufData.fromConnectSchema(schema);
      final Object deserialized = deserializer.deserialize(value, protobufSchema);

      if (deserialized == null) {
        return SchemaAndValue.NULL;
      } else {
        if (deserialized instanceof Message) {
          return protobufData.toConnectData(protobufSchema, (Message) deserialized);
        }
        throw new DataException(String.format(
            "Unsupported type %s returned during deserialization of topic %s ",
            deserialized.getClass().getName(),
            topic
        ));
      }
    } catch (SerializationException e) {
      throw new DataException(String.format(
          "Failed to deserialize data for topic %s to Protobuf: ",
          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 shouldThrowExceptionWhenUsedWithNoArgConstructor1() {
    // Given
    final ProtobufNoSRConverter protobufNoSRConverter = new ProtobufNoSRConverter();

    // When
    protobufNoSRConverter.toConnectData("topic", "test".getBytes(StandardCharsets.UTF_8));
  }