SplunkEventWriter extends DoFn<KV<Integer, SplunkEvent>, SplunkWriteError> { static Builder newBuilder() { return new AutoValue_SplunkEventWriter.Builder(); }  @Setup void setup(); @ProcessElement void processElement(
      @Element KV<Integer, SplunkEvent> input,
      OutputReceiver<SplunkWriteError> receiver,
      BoundedWindow window,
      @StateId(BUFFER_STATE_NAME) BagState<SplunkEvent> bufferState,
      @StateId(COUNT_STATE_NAME) ValueState<Long> countState,
      @TimerId(TIME_ID_NAME) Timer timer); @OnTimer(TIME_ID_NAME) void onExpiry(
      OutputReceiver<SplunkWriteError> receiver,
      @StateId(BUFFER_STATE_NAME) BagState<SplunkEvent> bufferState,
      @StateId(COUNT_STATE_NAME) ValueState<Long> countState); @Teardown void tearDown();  }
@Test public void eventWriterMissingURL() { Exception thrown = assertThrows(NullPointerException.class, () -> SplunkEventWriter.newBuilder().build()); assertTrue(thrown.getMessage().contains("url needs to be provided")); }
@Test public void eventWriterMissingToken() { Exception thrown = assertThrows( NullPointerException.class, () -> SplunkEventWriter.newBuilder().withUrl("test-url").build()); assertTrue(thrown.getMessage().contains("token needs to be provided")); }
@Test @Category(NeedsRunner.class) public void successfulSplunkWriteSingleBatchTest() { mockServerListening(200); int testPort = mockServerRule.getPort(); List<KV<Integer, SplunkEvent>> testEvents = ImmutableList.of( KV.of( 123, SplunkEvent.newBuilder() .withEvent("test-event-1") .withHost("test-host-1") .withIndex("test-index-1") .withSource("test-source-1") .withSourceType("test-source-type-1") .withTime(12345L) .create()), KV.of( 123, SplunkEvent.newBuilder() .withEvent("test-event-2") .withHost("test-host-2") .withIndex("test-index-2") .withSource("test-source-2") .withSourceType("test-source-type-2") .withTime(12345L) .create())); PCollection<SplunkWriteError> actual = pipeline .apply("Create Input data", Create.of(testEvents)) .apply( "SplunkEventWriter", ParDo.of( SplunkEventWriter.newBuilder() .withUrl(Joiner.on(':').join("http: .withInputBatchCount( StaticValueProvider.of(1)) .withToken("test-token") .build())); PAssert.that(actual).empty(); pipeline.run(); mockServerClient.verify( HttpRequest.request(EXPECTED_PATH), VerificationTimes.exactly(testEvents.size())); }
@Test @Category(NeedsRunner.class) public void successfulSplunkWriteMultiBatchTest() { mockServerListening(200); int testPort = mockServerRule.getPort(); List<KV<Integer, SplunkEvent>> testEvents = ImmutableList.of( KV.of( 123, SplunkEvent.newBuilder() .withEvent("test-event-1") .withHost("test-host-1") .withIndex("test-index-1") .withSource("test-source-1") .withSourceType("test-source-type-1") .withTime(12345L) .create()), KV.of( 123, SplunkEvent.newBuilder() .withEvent("test-event-2") .withHost("test-host-2") .withIndex("test-index-2") .withSource("test-source-2") .withSourceType("test-source-type-2") .withTime(12345L) .create())); PCollection<SplunkWriteError> actual = pipeline .apply("Create Input data", Create.of(testEvents)) .apply( "SplunkEventWriter", ParDo.of( SplunkEventWriter.newBuilder() .withUrl(Joiner.on(':').join("http: .withInputBatchCount( StaticValueProvider.of( testEvents.size())) .withToken("test-token") .build())); PAssert.that(actual).empty(); pipeline.run(); mockServerClient.verify(HttpRequest.request(EXPECTED_PATH), VerificationTimes.once()); }
@Test @Category(NeedsRunner.class) public void failedSplunkWriteSingleBatchTest() { mockServerListening(404); int testPort = mockServerRule.getPort(); List<KV<Integer, SplunkEvent>> testEvents = ImmutableList.of( KV.of( 123, SplunkEvent.newBuilder() .withEvent("test-event-1") .withHost("test-host-1") .withIndex("test-index-1") .withSource("test-source-1") .withSourceType("test-source-type-1") .withTime(12345L) .create())); PCollection<SplunkWriteError> actual = pipeline .apply("Create Input data", Create.of(testEvents)) .apply( "SplunkEventWriter", ParDo.of( SplunkEventWriter.newBuilder() .withUrl(Joiner.on(':').join("http: .withInputBatchCount( StaticValueProvider.of( testEvents.size())) .withToken("test-token") .build())); PAssert.that(actual) .containsInAnyOrder( SplunkWriteError.newBuilder() .withStatusCode(404) .withStatusMessage("Not Found") .withPayload( "{\"time\":12345,\"host\":\"test-host-1\"," + "\"source\":\"test-source-1\",\"sourcetype\":\"test-source-type-1\"," + "\"index\":\"test-index-1\",\"event\":\"test-event-1\"}") .create()); pipeline.run(); mockServerClient.verify(HttpRequest.request(EXPECTED_PATH), VerificationTimes.once()); }