@Test(groups = "fast") public void testWriteIOFailure() throws Exception { final DiskSpoolEventWriter writer = createWriter(writerSucceeds); try { writer.write(eventThrowsOnWrite); Assert.fail("expected IOException"); } catch (IOException e) { Assert.assertEquals(e.getClass(), IOException.class); } testSpoolDirs(0, 1, 0); }
@Override public synchronized void write(final Event event) throws IOException { if (!acceptsEvents) { log.warn("Writer not ready, discarding event: {}", event); return; } if (currentOutputter == null) { currentOutputFile = new File(tmpSpoolDirectory, String.format("%d.bin", fileId.incrementAndGet())); final FileOutputStream outputStream = codec.getFileOutputStream(currentOutputFile); if (eventSerializer == null) { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize); } else { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize, eventSerializer); } } try { final long startTime = System.nanoTime(); currentOutputter.writeObject(event); writeTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); } catch (RuntimeException e) { eventSerializationFailures.incrementAndGet(); throw new IOException("unable to serialize event", e); } catch (IOException e) { eventSerializationFailures.incrementAndGet(); try { forceCommit(); } catch (IOException ignored) { } throw new IOException("unable to serialize event", e); } }
DiskSpoolEventWriter implements EventWriter { @Override public synchronized void write(final Event event) throws IOException { if (!acceptsEvents) { log.warn("Writer not ready, discarding event: {}", event); return; } if (currentOutputter == null) { currentOutputFile = new File(tmpSpoolDirectory, String.format("%d.bin", fileId.incrementAndGet())); final FileOutputStream outputStream = codec.getFileOutputStream(currentOutputFile); if (eventSerializer == null) { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize); } else { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize, eventSerializer); } } try { final long startTime = System.nanoTime(); currentOutputter.writeObject(event); writeTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); } catch (RuntimeException e) { eventSerializationFailures.incrementAndGet(); throw new IOException("unable to serialize event", e); } catch (IOException e) { eventSerializationFailures.incrementAndGet(); try { forceCommit(); } catch (IOException ignored) { } throw new IOException("unable to serialize event", e); } } }
DiskSpoolEventWriter implements EventWriter { @Override public synchronized void write(final Event event) throws IOException { if (!acceptsEvents) { log.warn("Writer not ready, discarding event: {}", event); return; } if (currentOutputter == null) { currentOutputFile = new File(tmpSpoolDirectory, String.format("%d.bin", fileId.incrementAndGet())); final FileOutputStream outputStream = codec.getFileOutputStream(currentOutputFile); if (eventSerializer == null) { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize); } else { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize, eventSerializer); } } try { final long startTime = System.nanoTime(); currentOutputter.writeObject(event); writeTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); } catch (RuntimeException e) { eventSerializationFailures.incrementAndGet(); throw new IOException("unable to serialize event", e); } catch (IOException e) { eventSerializationFailures.incrementAndGet(); try { forceCommit(); } catch (IOException ignored) { } throw new IOException("unable to serialize event", e); } } DiskSpoolEventWriter(
        final EventHandler eventHandler,
        final String spoolPath,
        final boolean flushEnabled,
        final long flushIntervalInSeconds,
        final ScheduledExecutorService executor,
        final SyncType syncType,
        final int syncBatchSize
    );  DiskSpoolEventWriter(
        final EventHandler eventHandler,
        final String spoolPath,
        final boolean flushEnabled,
        final long flushIntervalInSeconds,
        final ScheduledExecutorService executor,
        final SyncType syncType,
        final int syncBatchSize,
        final CompressionCodec codec,
        final EventSerializer eventSerializer
    ); }
DiskSpoolEventWriter implements EventWriter { @Override public synchronized void write(final Event event) throws IOException { if (!acceptsEvents) { log.warn("Writer not ready, discarding event: {}", event); return; } if (currentOutputter == null) { currentOutputFile = new File(tmpSpoolDirectory, String.format("%d.bin", fileId.incrementAndGet())); final FileOutputStream outputStream = codec.getFileOutputStream(currentOutputFile); if (eventSerializer == null) { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize); } else { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize, eventSerializer); } } try { final long startTime = System.nanoTime(); currentOutputter.writeObject(event); writeTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); } catch (RuntimeException e) { eventSerializationFailures.incrementAndGet(); throw new IOException("unable to serialize event", e); } catch (IOException e) { eventSerializationFailures.incrementAndGet(); try { forceCommit(); } catch (IOException ignored) { } throw new IOException("unable to serialize event", e); } } DiskSpoolEventWriter(
        final EventHandler eventHandler,
        final String spoolPath,
        final boolean flushEnabled,
        final long flushIntervalInSeconds,
        final ScheduledExecutorService executor,
        final SyncType syncType,
        final int syncBatchSize
    );  DiskSpoolEventWriter(
        final EventHandler eventHandler,
        final String spoolPath,
        final boolean flushEnabled,
        final long flushIntervalInSeconds,
        final ScheduledExecutorService executor,
        final SyncType syncType,
        final int syncBatchSize,
        final CompressionCodec codec,
        final EventSerializer eventSerializer
    ); @Override synchronized void write(final Event event); @Override synchronized void commit(); @Override synchronized void forceCommit(); @Override synchronized void rollback(); @Override synchronized void close(); @Override String getSpoolPath(); @Managed(description = "Flush events (forward them to final handler)") void flush(); @Managed(description = "enable/disable flushing to hdfs") void setFlushEnabled(final boolean enabled); @Managed(description = "check if hdfs flushing is enabled") boolean getFlushEnabled(); @Managed(description = "set the commit interval for next scheduled commit to hdfs in seconds") void setFlushIntervalInSeconds(final long seconds); @Managed(description = "get the current commit interval to hdfs in seconds") long getFlushIntervalInSeconds(); @Managed(description = "size in kilobytes of disk spool queue not yet written to hdfs") long getDiskSpoolSize(); @Managed(description = "size in kilobytes of quarantined data that could not be written to hdfs") long getQuarantineSize(); @Managed(description = "attempt to process quarantined files") synchronized void processQuarantinedFiles(); @Managed(description = "count of events that could not be serialized from memory to disk") long getEventSeralizationFailureCount(); }
DiskSpoolEventWriter implements EventWriter { @Override public synchronized void write(final Event event) throws IOException { if (!acceptsEvents) { log.warn("Writer not ready, discarding event: {}", event); return; } if (currentOutputter == null) { currentOutputFile = new File(tmpSpoolDirectory, String.format("%d.bin", fileId.incrementAndGet())); final FileOutputStream outputStream = codec.getFileOutputStream(currentOutputFile); if (eventSerializer == null) { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize); } else { currentOutputter = ObjectOutputterFactory.createObjectOutputter(outputStream, syncType, syncBatchSize, eventSerializer); } } try { final long startTime = System.nanoTime(); currentOutputter.writeObject(event); writeTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); } catch (RuntimeException e) { eventSerializationFailures.incrementAndGet(); throw new IOException("unable to serialize event", e); } catch (IOException e) { eventSerializationFailures.incrementAndGet(); try { forceCommit(); } catch (IOException ignored) { } throw new IOException("unable to serialize event", e); } } DiskSpoolEventWriter(
        final EventHandler eventHandler,
        final String spoolPath,
        final boolean flushEnabled,
        final long flushIntervalInSeconds,
        final ScheduledExecutorService executor,
        final SyncType syncType,
        final int syncBatchSize
    );  DiskSpoolEventWriter(
        final EventHandler eventHandler,
        final String spoolPath,
        final boolean flushEnabled,
        final long flushIntervalInSeconds,
        final ScheduledExecutorService executor,
        final SyncType syncType,
        final int syncBatchSize,
        final CompressionCodec codec,
        final EventSerializer eventSerializer
    ); @Override synchronized void write(final Event event); @Override synchronized void commit(); @Override synchronized void forceCommit(); @Override synchronized void rollback(); @Override synchronized void close(); @Override String getSpoolPath(); @Managed(description = "Flush events (forward them to final handler)") void flush(); @Managed(description = "enable/disable flushing to hdfs") void setFlushEnabled(final boolean enabled); @Managed(description = "check if hdfs flushing is enabled") boolean getFlushEnabled(); @Managed(description = "set the commit interval for next scheduled commit to hdfs in seconds") void setFlushIntervalInSeconds(final long seconds); @Managed(description = "get the current commit interval to hdfs in seconds") long getFlushIntervalInSeconds(); @Managed(description = "size in kilobytes of disk spool queue not yet written to hdfs") long getDiskSpoolSize(); @Managed(description = "size in kilobytes of quarantined data that could not be written to hdfs") long getQuarantineSize(); @Managed(description = "attempt to process quarantined files") synchronized void processQuarantinedFiles(); @Managed(description = "count of events that could not be serialized from memory to disk") long getEventSeralizationFailureCount();  }