static YarnClusterSubmissionFromCS readYarnClusterSubmissionFromCSFromInputStream(
    final InputStream appInputStream, final InputStream jobInputStream) throws IOException {
  final JsonDecoder appDecoder = DecoderFactory.get().jsonDecoder(
      AvroYarnAppSubmissionParameters.getClassSchema(), appInputStream);
  final SpecificDatumReader<AvroYarnAppSubmissionParameters> appReader = new SpecificDatumReader<>(
      AvroYarnAppSubmissionParameters.class);
  final AvroYarnAppSubmissionParameters yarnClusterAppSubmissionParameters = appReader.read(null, appDecoder);

  final JsonDecoder jobDecoder = DecoderFactory.get().jsonDecoder(
      AvroYarnClusterJobSubmissionParameters.getClassSchema(), jobInputStream);
  final SpecificDatumReader<AvroYarnClusterJobSubmissionParameters> jobReader = new SpecificDatumReader<>(
      AvroYarnClusterJobSubmissionParameters.class);
  final AvroYarnClusterJobSubmissionParameters yarnClusterJobSubmissionParameters = jobReader.read(null, jobDecoder);

  return new YarnClusterSubmissionFromCS(yarnClusterAppSubmissionParameters, yarnClusterJobSubmissionParameters);
}
 

static GenericRecord deserializeJson(LegacyAvroSchema schema, String json) {
  String transformedJson = json;

  //if we had to apply any transforms to the schema to fix it (like escape illegal chars in identifiers)
  //we need to apply those same transformations to the json object to make it readable.
  for (SchemaTransformStep transform : schema.getTransforms()) {
    transformedJson = transform.applyToJsonObject(transformedJson);
  }

  List<PayloadTransformStep> stepsTaken = new ArrayList<>(1);
  while (true) {
    try {
      JsonDecoder decoder = AvroCompatibilityHelper.newJsonDecoder(schema.getFixedSchema(), transformedJson);
      DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema.getFixedSchema());
      return reader.read(null, decoder);
    } catch (Exception issue) {
      if (stepsTaken.size() > MAX_STEPS_ATTEMPTED) {
        throw new IllegalArgumentException("unable to deserialize json");
      }

      PayloadTransformStep step = findFixFor(schema, issue);
      if (step == null) {
        //if we got here we have no idea what the issue is nor how to fix it
        throw new IllegalStateException("unhandled", issue);
      }
      String fixedJson = step.applyToJsonPayload(transformedJson);
      if (fixedJson.equals(transformedJson)) {
        throw new IllegalStateException("made no progress fixing json");
      }
      transformedJson = fixedJson;
      stepsTaken.add(step);
    }
  }
}
 

/**
 * Convert json bytes back into avro record.
 */
public static GenericRecord jsonBytesToAvro(byte[] bytes, Schema schema) throws IOException {
  ByteArrayInputStream bio = new ByteArrayInputStream(bytes);
  JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(schema, bio);
  GenericDatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
  return reader.read(null, jsonDecoder);
}
 

@SuppressWarnings( {"rawtypes", "deprecation"})
@Override
public void parse(InputStream is) throws Exception {
    int eventCtr = 0;
    try {
        final long start = System.currentTimeMillis();
        DataInputStream in = new DataInputStream(is);
        String version = in.readLine();
        if (!"Avro-Json".equals(version)) {
            throw new IOException("Incompatible event log version: " + version);
        }

        Schema schema = Schema.parse(in.readLine());
        SpecificDatumReader datumReader = new SpecificDatumReader(schema);
        JsonDecoder decoder = DecoderFactory.get().jsonDecoder(schema, in);

        Event wrapper;
        while ((wrapper = getNextEvent(datumReader, decoder)) != null) {
            ++eventCtr;
            reader.handleEvent(wrapper);
        }
        reader.parseConfiguration();
        // don't need put to finally as it's a kind of flushing data
        reader.close();
        logger.info("reader used " + (System.currentTimeMillis() - start) + "ms");
    } catch (Exception ioe) {
        logger.error("Caught exception parsing history file after " + eventCtr + " events", ioe);
        throw ioe;
    } finally {
        if (is != null) {
            is.close();
        }
    }
}
 

/**
 * Reads avro object from input stream.
 *
 * @param inputStream The input stream to read from
 * @return Avro object
 * @throws IOException
 */
AvroYarnJobSubmissionParameters fromInputStream(final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
          AvroYarnJobSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroYarnJobSubmissionParameters> reader = new SpecificDatumReader<>(
          AvroYarnJobSubmissionParameters.class);
  return reader.read(null, decoder);
}
 

private static AvroAzureBatchJobSubmissionParameters readAvroJobSubmissionParameters(
    final File paramsFile) throws IOException {
  final AvroAzureBatchJobSubmissionParameters avroAzureBatchJobSubmissionParameters;
  try (FileInputStream fileInputStream = new FileInputStream(paramsFile)) {
    final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
        AvroAzureBatchJobSubmissionParameters.getClassSchema(), fileInputStream);
    final SpecificDatumReader<AvroAzureBatchJobSubmissionParameters> reader =
        new SpecificDatumReader<>(AvroAzureBatchJobSubmissionParameters.class);
    avroAzureBatchJobSubmissionParameters = reader.read(null, decoder);
  }
  return avroAzureBatchJobSubmissionParameters;
}
 

static AvroYarnAppSubmissionParameters readYarnAppSubmissionParametersFromInputStream(
    final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
      AvroYarnAppSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroYarnAppSubmissionParameters> reader = new SpecificDatumReader<>(
      AvroYarnAppSubmissionParameters.class);
  return reader.read(null, decoder);
}
 

static AvroYarnJobSubmissionParameters readYarnJobSubmissionParametersFromInputStream(
    final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
      AvroYarnJobSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroYarnJobSubmissionParameters> reader = new SpecificDatumReader<>(
      AvroYarnJobSubmissionParameters.class);
  return reader.read(null, decoder);
}
 

/**
 * Reads avro object from input stream.
 *
 * @param inputStream The input stream to read from
 * @return Avro object
 * @throws IOException
 */
AvroMultiRuntimeAppSubmissionParameters fromInputStream(final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
          AvroMultiRuntimeAppSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroMultiRuntimeAppSubmissionParameters> reader = new SpecificDatumReader<>(
          AvroMultiRuntimeAppSubmissionParameters.class);
  return reader.read(null, decoder);
}
 

private String getLastProcessedTimestamp() {

    String timestamp = "";
    try {
      final SubscriberStubSettings subscriberStubSettings =
          SubscriberStubSettings.newBuilder()
              .setTransportChannelProvider(
                  SubscriberStubSettings.defaultGrpcTransportProviderBuilder()
                      .setMaxInboundMessageSize(20 << 20) // 20MB
                      .build())
              .build();

      try (SubscriberStub subscriber = GrpcSubscriberStub.create(subscriberStubSettings)) {
        final String subscriptionName = ProjectSubscriptionName.format(PROJECT_ID, tableName);
        final PullRequest pullRequest =
            PullRequest.newBuilder()
                .setMaxMessages(1)
                .setReturnImmediately(true)
                .setSubscription(subscriptionName)
                .build();

        final PullResponse pullResponse = subscriber.pullCallable().call(pullRequest);
        final DatumReader<GenericRecord> datumReader =
            new GenericDatumReader<GenericRecord>(avroSchema);

        for (ReceivedMessage message : pullResponse.getReceivedMessagesList()) {
          final JsonDecoder decoder =
              DecoderFactory.get()
                  .jsonDecoder(avroSchema, message.getMessage().getData().newInput());

          final GenericRecord record = datumReader.read(null, decoder);
          timestamp = record.get("Timestamp").toString();

          log.debug("---------------- Got Timestamp: " + timestamp);
        }
      }
    } catch (IOException e) {
      log.error("Could not get last processed timestamp from pub / sub", e);

      // If we cannot find a previously processed timestamp, we will default
      // to the one present in the config file.
      return startingTimestamp;
    }

    return timestamp;
  }
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException {
  return new JsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException {
  return new JsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

@Override
public JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException {
  return DecoderFactory.get().jsonDecoder(schema, in);
}
 

public static JsonDecoder newJsonDecoder(Schema schema, InputStream input) throws IOException {
  return FACTORY.newJsonDecoder(schema, input);
}
 

public static JsonDecoder newJsonDecoder(Schema schema, String input) throws IOException {
  return FACTORY.newJsonDecoder(schema, input);
}
 

/**
 * constructs a {@link JsonDecoder} on top of the given {@link InputStream} for the given {@link Schema}
 * @param schema a schema
 * @param in an input stream
 * @return a decoder
 * @throws IOException on io errors
 */
public static JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException {
  assertAvroAvailable();
  return ADAPTER.newJsonDecoder(schema, in);
}
 

/**
 * constructs a {@link JsonDecoder} on top of the given {@link String} for the given {@link Schema}
 * @param schema a schema
 * @param in a String containing a json-serialized avro payload
 * @return a decoder
 * @throws IOException on io errors
 */
public static JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException {
  assertAvroAvailable();
  return ADAPTER.newJsonDecoder(schema, in);
}
 

/**
 * Decode and deserialize the Json byte array into an instance of an Avro record
 * @param schema schema describing the expected information of the bytes.
 * @param bytes Json string in bytes to decode
 * @return decoded instance of GenericRecord
 */
public static <T> T decodeJsonAsAvroGenericRecord(Schema schema, byte[] bytes, T reuse) throws IOException {
  JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(schema, new String(bytes, StandardCharsets.UTF_8));
  GenericDatumReader<T> reader = new GenericDatumReader<>(schema);
  return reader.read(reuse, jsonDecoder);
}
 

JsonDecoder newJsonDecoder(Schema schema, InputStream in) throws IOException; 

JsonDecoder newJsonDecoder(Schema schema, String in) throws IOException;